cmd, eth, server, pm: Streamflow client compatibility

cmd/devtool/devtool.go

@@ -237,7 +237,7 @@ func ethSetup(ethAcctAddr, keystoreDir string, isBroadcaster bool) {
 		}
 		glog.Info("Done initializing round.")
 		glog.Info("Activating transcoder")
-		// curl -d "blockRewardCut=10&feeShare=5&pricePerSegment=1&amount=500" --data-urlencode "serviceURI=https://$transcoderServiceAddr" \
+		// curl -d "blockRewardCut=10&feeShare=5&amount=500" --data-urlencode "serviceURI=https://$transcoderServiceAddr" \
 		//   -H "Content-Type: application/x-www-form-urlencoded" \
 		//   -X "POST" http://localhost:$transcoderCliPort/activateTranscoder\
 		var amount *big.Int = big.NewInt(int64(500))
@@ -256,9 +256,8 @@ func ethSetup(ethAcctAddr, keystoreDir string, isBroadcaster bool) {
 			return
 		}
 		glog.Infof("Registering transcoder %v", ethAcctAddr)
-		price := big.NewInt(1)
 
-		tx, err = client.Transcoder(eth.FromPerc(10), eth.FromPerc(5), price)
+		tx, err = client.Transcoder(eth.FromPerc(10), eth.FromPerc(5))
 		if err == eth.ErrCurrentRoundLocked {
 			// wait for next round and retry
 		}
@@ -314,7 +313,7 @@ func createRunScript(ethAcctAddr, dataDir, serviceHost string, isBroadcaster boo
 		script += fmt.Sprintf(` -initializeRound=true \
     -serviceAddr %s:%d  -transcoder=true -orchestrator=true \
     -orchSecret secre -pricePerUnit 1
-    `, serviceHost, mediaPort, dataDir)
+    `, serviceHost, mediaPort)
 	} else {
 		script += fmt.Sprintf(` -broadcaster=true -rtmpAddr %s:%d`, serviceHost, rtmpPort)
 	}

cmd/livepeer/livepeer.go

@@ -383,7 +383,7 @@ func main() {
 		go unbondingWatcher.Watch()
 		defer unbondingWatcher.Stop()
 
-		senderWatcher, err := watchers.NewSenderWatcher(addrMap["TicketBroker"], blockWatcher, n.Eth)
+		senderWatcher, err := watchers.NewSenderWatcher(addrMap["TicketBroker"], blockWatcher, n.Eth, roundsWatcher)
 		if err != nil {
 			glog.Errorf("Failed to setup senderwatcher: %v", err)
 			return

cmd/livepeer_cli/wizard_bond.go

@@ -34,7 +34,7 @@ func (w *wizard) registeredOrchestratorStats() map[int]common.Address {
 	fmt.Println("+------------------------+")
 
 	table := tablewriter.NewWriter(os.Stdout)
-	table.SetHeader([]string{"ID", "Address", "Active", "Delegated Stake", "Reward Cut (%)", "Fee Share (%)", "Price", "Pending Reward Cut (%)", "Pending Fee Share (%)", "Pending Price", "Service URI"})
+	table.SetHeader([]string{"ID", "Address", "Active", "Delegated Stake", "Reward Cut (%)", "Fee Share (%)", "Service URI"})
 
 	for _, t := range orchestrators {
 		table.Append([]string{
@@ -44,10 +44,6 @@ func (w *wizard) registeredOrchestratorStats() map[int]common.Address {
 			eth.FormatUnits(t.DelegatedStake, "LPT"),
 			eth.FormatPerc(t.RewardCut),
 			eth.FormatPerc(t.FeeShare),
-			eth.FormatUnits(t.PricePerSegment, "ETH"),
-			eth.FormatPerc(t.PendingRewardCut),
-			eth.FormatPerc(t.PendingFeeShare),
-			eth.FormatUnits(t.PendingPricePerSegment, "ETH"),
 			t.ServiceURI,
 		})
 
@@ -337,7 +333,12 @@ func (w *wizard) withdrawFees() {
 }
 
 func (w *wizard) claimRewardsAndFees() {
-	fmt.Printf("Current round: %v\n", w.currentRound())
+	currentRound, err := w.currentRound()
+	if err != nil {
+		glog.Errorf("error getting current round: %v\n", err)
+		return
+	}
+	fmt.Printf("Current round: %v\n", currentRound.Int64())
 
 	d, err := w.getDelegatorInfo()
 	if err != nil {

cmd/livepeer_cli/wizard_rounds.go

@@ -1,11 +1,31 @@
 package main
 
 import (
+	"errors"
 	"fmt"
+	"io/ioutil"
+	"math/big"
+	"net/http"
 )
 
-func (w *wizard) currentRound() string {
-	return httpGet(fmt.Sprintf("http://%v:%v/currentRound", w.host, w.httpPort))
+func (w *wizard) currentRound() (*big.Int, error) {
+	resp, err := http.Get(fmt.Sprintf("http://%v:%v/currentRound", w.host, w.httpPort))
+	if err != nil {
+		return nil, err
+	}
+
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, errors.New("http response status not ok")
+	}
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	return new(big.Int).SetBytes(body), nil
 }
 
 func (w *wizard) initializeRound() {

cmd/livepeer_cli/wizard_stats.go

@@ -15,7 +15,6 @@ import (
 	"github.com/livepeer/go-livepeer/eth"
 	lpTypes "github.com/livepeer/go-livepeer/eth/types"
 	"github.com/livepeer/go-livepeer/net"
-	"github.com/livepeer/go-livepeer/pm"
 	"github.com/olekukonko/tablewriter"
 )
 
@@ -76,7 +75,11 @@ func (w *wizard) stats(showOrchestrator bool) {
 		w.delegatorStats()
 	}
 
-	currentRound := w.currentRound()
+	currentRound, err := w.currentRound()
+	if err != nil {
+		glog.Errorf("Error getting current round: %v", err)
+		return
+	}
 
 	fmt.Printf("CURRENT ROUND: %v\n", currentRound)
 }
@@ -153,19 +156,15 @@ func (w *wizard) broadcastStats() {
 		[]string{"Max Price Per Pixel", priceString},
 		[]string{"Broadcast Transcoding Options", transcodingOptions},
 		[]string{"Deposit", eth.FormatUnits(sender.Deposit, "ETH")},
-		[]string{"Reserve", eth.FormatUnits(sender.Reserve, "ETH")},
+		[]string{"Reserve", eth.FormatUnits(sender.Reserve.FundsRemaining, "ETH")},
 	}
 
 	for _, v := range data {
 		table.Append(v)
 	}
 
-	if sender.ReserveState != pm.NotFrozen {
-		table.Append([]string{"Thaw Round", sender.ThawRound.String()})
-	}
-
-	if sender.WithdrawBlock.Cmp(big.NewInt(0)) > 0 && (sender.Deposit.Cmp(big.NewInt(0)) > 0 || sender.Reserve.Cmp(big.NewInt(0)) > 0) {
-		table.Append([]string{"Withdraw Block", sender.WithdrawBlock.String()})
+	if sender.WithdrawRound.Cmp(big.NewInt(0)) > 0 && (sender.Deposit.Cmp(big.NewInt(0)) > 0 || sender.Reserve.FundsRemaining.Cmp(big.NewInt(0)) > 0) {
+		table.Append([]string{"Withdraw Round", sender.WithdrawRound.String()})
 	}
 
 	table.SetAlignment(tablewriter.ALIGN_RIGHT)
@@ -197,8 +196,6 @@ func (w *wizard) orchestratorStats() {
 		[]string{"Delegated Stake", eth.FormatUnits(t.DelegatedStake, "LPT")},
 		[]string{"Reward Cut (%)", eth.FormatPerc(t.RewardCut)},
 		[]string{"Fee Share (%)", eth.FormatPerc(t.FeeShare)},
-		[]string{"Pending Reward Cut (%)", eth.FormatPerc(t.PendingRewardCut)},
-		[]string{"Pending Fee Share (%)", eth.FormatPerc(t.PendingFeeShare)},
 		[]string{"Last Reward Round", t.LastRewardRound.String()},
 		[]string{"Base price per pixel", fmt.Sprintf("%v wei / %v pixels", priceInfo.Num(), priceInfo.Denom())},
 	}

cmd/livepeer_cli/wizard_ticketbroker.go

@@ -47,17 +47,7 @@ func (w *wizard) deposit() {
 	}
 
 	fmt.Printf("Current Deposit: %v\n", sender.Deposit)
-	fmt.Printf("Current Reserve: %v\n", sender.Reserve)
-
-	if sender.ReserveState == pm.Frozen {
-		currRound := w.currentRound()
-
-		fmt.Printf("Current Round: %v\n", currRound)
-		fmt.Printf("Thaw Round: %v\n", sender.ThawRound)
-
-		fmt.Printf("Cannot deposit because sender's reserve is frozen and not yet thawed")
-		return
-	}
+	fmt.Printf("Current Reserve: %v\n", sender.Reserve.FundsRemaining)
 
 	fmt.Printf("Enter deposit amount in ETH - ")
 
@@ -83,28 +73,18 @@ func (w *wizard) unlock() {
 		return
 	}
 
-	blk, err := w.currentBlock()
+	round, err := w.currentRound()
 	if err != nil {
-		glog.Errorf("Error getting current block: %v", err)
+		glog.Errorf("Error getting current round: %v", err)
 		return
 	}
 
 	fmt.Printf("Current Deposit: %v\n", eth.FormatUnits(sender.Deposit, "ETH"))
-	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve, "ETH"))
-
-	if sender.ReserveState == pm.Frozen {
-		currRound := w.currentRound()
-
-		fmt.Printf("Current Round: %v\n", currRound)
-		fmt.Printf("Thaw Round: %v\n", sender.ThawRound)
-
-		fmt.Printf("Cannot deposit because sender's reserve is frozen and not yet thawed")
-		return
-	}
+	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve.FundsRemaining, "ETH"))
 
-	fmt.Printf("Current Block: %v\n", blk)
+	fmt.Printf("Current Round: %v\n", round)
 
-	ss := senderStatus(sender, blk)
+	ss := senderStatus(sender, round)
 	if ss != Locked {
 		printSenderStatus("Cannot unlock because sender's deposit and reserve are not locked", ss)
 		return
@@ -116,9 +96,9 @@ func (w *wizard) unlock() {
 		return
 	}
 
-	projWithdrawBlock := new(big.Int).Add(blk, params.UnlockPeriod)
+	projWithdrawRound := new(big.Int).Add(round, params.UnlockPeriod)
 
-	fmt.Printf("If you initiate the unlock period now, you will be able to withdraw at block %v\n", projWithdrawBlock)
+	fmt.Printf("If you initiate the unlock period now, you will be able to withdraw at round %v\n", projWithdrawRound)
 	fmt.Printf("Would you like initiate the unlock period? (y/n) - ")
 
 	input := w.readStringYesOrNo()
@@ -136,18 +116,18 @@ func (w *wizard) cancelUnlock() {
 		return
 	}
 
-	blk, err := w.currentBlock()
+	round, err := w.currentRound()
 	if err != nil {
-		glog.Errorf("Error getting current block: %v", err)
+		glog.Errorf("Error getting current round: %v", err)
 		return
 	}
 
 	fmt.Printf("Current Deposit: %v\n", eth.FormatUnits(sender.Deposit, "ETH"))
-	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve, "ETH"))
-	fmt.Printf("Current Block: %v\n", blk)
-	fmt.Printf("Withdraw Block: %v\n", sender.WithdrawBlock)
+	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve.FundsRemaining, "ETH"))
+	fmt.Printf("Current Round: %v\n", round)
+	fmt.Printf("Withdraw Round: %v\n", sender.WithdrawRound)
 
-	ss := senderStatus(sender, blk)
+	ss := senderStatus(sender, round)
 	if ss != Unlocking {
 		printSenderStatus("Cannot cancel unlock because sender is not in the unlock period", ss)
 		return
@@ -171,33 +151,21 @@ func (w *wizard) withdraw() {
 	}
 
 	fmt.Printf("Current Deposit: %v\n", eth.FormatUnits(sender.Deposit, "ETH"))
-	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve, "ETH"))
+	fmt.Printf("Current Reserve: %v\n", eth.FormatUnits(sender.Reserve.FundsRemaining, "ETH"))
 
-	if sender.ReserveState != pm.NotFrozen {
-		currRound := w.currentRound()
-
-		fmt.Printf("Current Round: %v\n", currRound)
-		fmt.Printf("Thaw Round: %v\n", sender.ThawRound)
-
-		if sender.ReserveState == pm.Frozen {
-			fmt.Printf("Cannot withdraw because sender's reserve is frozen and not yet thawed")
-			return
-		}
-	} else {
-		blk, err := w.currentBlock()
-		if err != nil {
-			glog.Errorf("Error getting current block: %v", err)
-			return
-		}
+	round, err := w.currentRound()
+	if err != nil {
+		glog.Errorf("Error getting current round: %v", err)
+		return
+	}
 
-		fmt.Printf("Current Block: %v\n", blk)
-		fmt.Printf("Withdraw Block: %v\n", sender.WithdrawBlock)
+	fmt.Printf("Current Round: %v\n", round)
+	fmt.Printf("Withdraw Round: %v\n", sender.WithdrawRound)
 
-		ss := senderStatus(sender, blk)
-		if ss != Unlocked {
-			printSenderStatus("Cannot withdraw because sender's deposit and reserve are not unlocked", ss)
-			return
-		}
+	ss := senderStatus(sender, round)
+	if ss != Unlocked {
+		printSenderStatus("Cannot withdraw because sender's deposit and reserve are not unlocked", ss)
+		return
 	}
 
 	fmt.Printf("Would you like to withdraw? (y/n) - ")
@@ -220,9 +188,11 @@ func (w *wizard) senderInfo() (info pm.SenderInfo, err error) {
 	if resp.StatusCode == http.StatusInternalServerError {
 		// node is in offchain mode
 		info.Deposit = big.NewInt(0)
-		info.Reserve = big.NewInt(0)
-		info.ThawRound = big.NewInt(0)
-		info.WithdrawBlock = big.NewInt(0)
+		info.WithdrawRound = big.NewInt(0)
+		info.Reserve = &pm.ReserveInfo{
+			FundsRemaining:        big.NewInt(0),
+			ClaimedInCurrentRound: big.NewInt(0),
+		}
 		return
 	}
 
@@ -282,13 +252,13 @@ func printSenderStatus(msg string, status SenderStatus) {
 	fmt.Printf("%v: %v\n", msg, statusMsg)
 }
 
-func senderStatus(sender pm.SenderInfo, currentBlock *big.Int) SenderStatus {
-	if sender.Deposit.Cmp(big.NewInt(0)) == 0 && sender.Reserve.Cmp(big.NewInt(0)) == 0 {
+func senderStatus(sender pm.SenderInfo, currentRound *big.Int) SenderStatus {
+	if sender.Deposit.Cmp(big.NewInt(0)) == 0 && sender.Reserve.FundsRemaining.Cmp(big.NewInt(0)) == 0 {
 		return Empty
 	}
 
-	if sender.WithdrawBlock.Cmp(big.NewInt(0)) > 0 {
-		if sender.WithdrawBlock.Cmp(currentBlock) <= 0 {
+	if sender.WithdrawRound.Cmp(big.NewInt(0)) > 0 {
+		if sender.WithdrawRound.Cmp(currentRound) <= 0 {
 			return Unlocked
 		}
 

cmd/livepeer_cli/wizard_ticketbroker_test.go

@@ -8,10 +8,13 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func createSender(deposit *big.Int, reserve *big.Int, withdrawBlock *big.Int) (sender pm.SenderInfo) {
+func createSender(deposit *big.Int, reserve *big.Int, withdrawRound *big.Int) (sender pm.SenderInfo) {
 	sender.Deposit = deposit
-	sender.Reserve = reserve
-	sender.WithdrawBlock = withdrawBlock
+	sender.WithdrawRound = withdrawRound
+	sender.Reserve = &pm.ReserveInfo{
+		FundsRemaining:        reserve,
+		ClaimedInCurrentRound: big.NewInt(0),
+	}
 
 	return
 }
@@ -24,17 +27,17 @@ func TestSenderStatus(t *testing.T) {
 	ss := senderStatus(s, big.NewInt(0))
 	assert.Equal(Empty, ss)
 
-	// Test Empty, but withdrawBlock > 0
+	// Test Empty, but WithdrawRound > 0
 	s = createSender(big.NewInt(0), big.NewInt(0), big.NewInt(5))
 	ss = senderStatus(s, big.NewInt(0))
 	assert.Equal(Empty, ss)
 
-	// Test Unlocked when withdrawBlock = currentBlock
+	// Test Unlocked when WithdrawRound = currentRound
 	s = createSender(big.NewInt(7), big.NewInt(0), big.NewInt(5))
 	ss = senderStatus(s, big.NewInt(5))
 	assert.Equal(Unlocked, ss)
 
-	// Test Unlocked when withdrawBlock < currentBlock
+	// Test Unlocked when WithdrawRound < currentRound
 	s = createSender(big.NewInt(7), big.NewInt(0), big.NewInt(5))
 	ss = senderStatus(s, big.NewInt(6))
 	assert.Equal(Unlocked, ss)

cmd/livepeer_cli/wizard_transcoder.go

@@ -172,12 +172,12 @@ func (w *wizard) callReward() {
 		fmt.Printf("Error getting orchestrator info: %v\n", err)
 		return
 	}
-	c, err := strconv.ParseInt(w.currentRound(), 10, 64)
+	c, err := w.currentRound()
 	if err != nil {
 		fmt.Printf("Error converting current round: %v\n", c)
 	}
 
-	if c == t.LastRewardRound.Int64() {
+	if c.Cmp(t.LastRewardRound) == 0 {
 		fmt.Printf("Reward for current round %v already called\n", c)
 		return
 	}