Merge pull request #22 from spacemeshos/account-streams

Account streams

.gitignore

@@ -7,4 +7,4 @@ cli_wallet*
 wallet
 myWallet*Z.json
 *.json
-
+log.*

AUTHORS

@@ -1,11 +1,8 @@
-cli-wallet was created by Aviv Eyal (avive@spacemesh.io)
-
-Contributions from 
-
 Moshe Shababo
 Anton Lerner
 Noam Nelke
 Alexey Sudachen
 rehs0y
+Dave Appleton
+Aviv Eyal
 
-New Wallet format support was added by Dave Appleton (calistralabs@gmail.com) in January 2021

client/global_state_service.go

@@ -74,6 +74,28 @@ func (c *gRPCClient) AccountRewards(address gosmtypes.Address, offset uint32, ma
 	return rewards, resp.TotalResults, nil
 }
 
+// getAccountStream returns an AccountDataStreamClient
+func (c *gRPCClient) getAccountStream(address gosmtypes.Address, flag apitypes.AccountDataFlag) (apitypes.GlobalStateService_AccountDataStreamClient, error) {
+	gsc := c.getGlobalStateServiceClient()
+	return gsc.AccountDataStream(context.Background(), &apitypes.AccountDataStreamRequest{
+		Filter: &apitypes.AccountDataFilter{
+			AccountId: &apitypes.AccountId{
+				Address: address.Bytes()},
+			AccountDataFlags: uint32(flag),
+		},
+	})
+}
+
+// AccountRewardsStream returns a stream of account rewards
+func (c *gRPCClient) AccountRewardsStream(address gosmtypes.Address) (apitypes.GlobalStateService_AccountDataStreamClient, error) {
+	return c.getAccountStream(address, apitypes.AccountDataFlag_ACCOUNT_DATA_FLAG_REWARD)
+}
+
+// AccountRewardsStream returns a stream of account changes
+func (c *gRPCClient) AccountUpdatesStream(address gosmtypes.Address) (apitypes.GlobalStateService_AccountDataStreamClient, error) {
+	return c.getAccountStream(address, apitypes.AccountDataFlag_ACCOUNT_DATA_FLAG_ACCOUNT)
+}
+
 // AccountTransactionsReceipts returns transaction receipts for an account
 func (c *gRPCClient) AccountTransactionsReceipts(address gosmtypes.Address, offset uint32, maxResults uint32) ([]*apitypes.TransactionReceipt, uint32, error) {
 	gsc := c.getGlobalStateServiceClient()

client/smesher_service.go

@@ -49,7 +49,7 @@ func (c *gRPCClient) GetPostComputeProviders() ([]*apitypes.PostComputeProvider,
 	}
 }
 
-// GetPostComputeProviders returns the proof of space generators available on the system
+// CreatePostData starts or continues pos data creation operation
 func (c *gRPCClient) CreatePostData(data *apitypes.PostData) (*status.Status, error) {
 	s := c.getSmesherServiceClient()
 	if resp, err := s.CreatePostData(context.Background(), &apitypes.CreatePostDataRequest{Data: data}); err != nil {
@@ -84,7 +84,7 @@ func (c *gRPCClient) StopSmeshing(deleteFiles bool) (*status.Status, error) {
 	return resp.Status, nil
 }
 
-// SetCoinbase sets the smesher's coinbase address
+// SetRewardsAddress sets the smesher's rewards address
 func (c *gRPCClient) SetRewardsAddress(address gosmtypes.Address) (*status.Status, error) {
 	s := c.getSmesherServiceClient()
 	resp, err := s.SetCoinbase(context.Background(), &apitypes.SetCoinbaseRequest{Id: &apitypes.AccountId{Address: address.Bytes()}})
@@ -94,7 +94,7 @@ func (c *gRPCClient) SetRewardsAddress(address gosmtypes.Address) (*status.Statu
 	return resp.Status, nil
 }
 
-// GetCoinbase get the smesher's current rewards address
+// GetRewardsAddress get the smesher's current rewards address
 func (c *gRPCClient) GetRewardsAddress() (*gosmtypes.Address, error) {
 	s := c.getSmesherServiceClient()
 	resp, err := s.Coinbase(context.Background(), &empty.Empty{})

log/log.go

@@ -89,8 +89,8 @@ func getBackendLevelWithFileBackend(module, prefix, logFormat, fileFormat, dataF
 	return leveledBackends
 }
 
-// initSpacemeshLoggingSystem initializes app logging system.
-func initSpacemeshLoggingSystem(dataFolderPath string, logFileName string) {
+// InitSpacemeshLoggingSystem initializes app logging system.
+func InitSpacemeshLoggingSystem(dataFolderPath string, logFileName string) {
 	log := logging.MustGetLogger("app")
 
 	// we wrap all log calls so we need to add 1 to call depth

repl/account_commands.go

@@ -109,7 +109,7 @@ func coinAmount(val uint64) string {
 	}
 }
 
-// printAccountInfo prints current wallet account info from global state
+// printAccountInfo prints current wallet's account info from global state
 func (r *repl) printAccountInfo() {
 	acc, err := r.getCurrent()
 	if err != nil {
@@ -118,29 +118,14 @@ func (r *repl) printAccountInfo() {
 	}
 
 	address := gosmtypes.BytesToAddress(acc.PubKey)
-	state, err := r.client.AccountState(address)
+	account, err := r.client.AccountState(address)
 	if err != nil {
 		log.Error("failed to get account info: %v", err)
 		return
 	}
 
-	currBalance := uint64(0)
-	if state.StateCurrent.Balance != nil {
-		currBalance = state.StateCurrent.Balance.Value
-	}
-
-	projectedBalance := uint64(0)
-	if state.StateProjected.Balance != nil {
-		projectedBalance = state.StateProjected.Balance.Value
-	}
-
 	fmt.Println(printPrefix, "Local alias:", acc.Name)
-	fmt.Println(printPrefix, "Address:", address.String())
-	fmt.Println(printPrefix, "Balance:", coinAmount(currBalance)) // currBalance, coinUnitName)
-	fmt.Println(printPrefix, "Nonce:", state.StateCurrent.Counter)
-	fmt.Println(printPrefix, "Projected Balance:", coinAmount(projectedBalance)) // projectedBalance, coinUnitName)
-	fmt.Println(printPrefix, "Projected Nonce:", state.StateProjected.Counter)
-	fmt.Println(printPrefix, "Projected account state includes all pending transactions that haven't been added to the mesh yet.")
+	printAccount(account, address)
 	fmt.Println(printPrefix, fmt.Sprintf("Public key: 0x%s", hex.EncodeToString(acc.PubKey)))
 	fmt.Println(printPrefix, fmt.Sprintf("Private key: 0x%s", hex.EncodeToString(acc.PrivKey)))
 }
@@ -155,6 +140,26 @@ func (r *repl) printLocalAccountRewards() {
 	r.printRewards(acc.Address())
 }
 
+// printAccountState prints the account data member
+func printAccount(account *apitypes.Account, address gosmtypes.Address) {
+	currBalance := uint64(0)
+	if account.StateCurrent.Balance != nil {
+		currBalance = account.StateCurrent.Balance.Value
+	}
+
+	projectedBalance := uint64(0)
+	if account.StateProjected.Balance != nil {
+		projectedBalance = account.StateProjected.Balance.Value
+	}
+
+	fmt.Println(printPrefix, "Address:", address.String())
+	fmt.Println(printPrefix, "Balance:", coinAmount(currBalance)) // currBalance, coinUnitName)
+	fmt.Println(printPrefix, "Nonce:", account.StateCurrent.Counter)
+	fmt.Println(printPrefix, "Projected Balance:", coinAmount(projectedBalance)) // projectedBalance, coinUnitName)
+	fmt.Println(printPrefix, "Projected Nonce:", account.StateProjected.Counter)
+	fmt.Println(printPrefix, "Projected state includes all pending transactions that haven't been added to the mesh yet.")
+}
+
 // printReward prints a Reward
 func printReward(r *apitypes.Reward) {
 	fmt.Println(printPrefix, "Rewarded on layer:", r.Layer.Number)

repl/global_state_commands.go

@@ -3,6 +3,7 @@ package repl
 import (
 	"encoding/hex"
 	"fmt"
+	"io"
 
 	gosmtypes "github.com/spacemeshos/go-spacemesh/common/types"
 	"github.com/spacemeshos/go-spacemesh/common/util"
@@ -33,6 +34,68 @@ func (r *repl) printAccountRewards() {
 	r.printRewards(addr)
 }
 
+// printAccountRewardsStream prints new rewards awarded to an account
+func (r *repl) printAccountRewardsStream() {
+	addrStr := inputNotBlank(enterAddressMsg)
+	addr := gosmtypes.HexToAddress(addrStr)
+	streamClient, err := r.client.AccountRewardsStream(addr)
+	if err != nil {
+		log.Error("failed to get rewards stream for account: %v", err)
+		return
+	}
+
+	fmt.Println(printPrefix, "Listening to new rewards for address: ", addr.String())
+
+	done := make(chan bool)
+	go func() {
+		for {
+			resp, err := streamClient.Recv()
+			if err == io.EOF {
+				// server closed the stream
+				log.Info("api server closed the server-side stream")
+				done <- true
+			} else if err != nil {
+				log.Error("error reading from rewards stream: %v", err)
+				done <- true
+			}
+
+			reward := resp.GetDatum().GetReward()
+			printReward(reward)
+		}
+	}()
+}
+
+// printAccountRewardsStream prints account state updates
+func (r *repl) printAccountUpdatesStream() {
+	addrStr := inputNotBlank(enterAddressMsg)
+	address := gosmtypes.HexToAddress(addrStr)
+	streamClient, err := r.client.AccountRewardsStream(address)
+	if err != nil {
+		log.Error("failed to get updates stream for account: %v", err)
+		return
+	}
+
+	fmt.Println(printPrefix, "Listening for new updates for address: ", address.String())
+
+	done := make(chan bool)
+	go func() {
+		for {
+			resp, err := streamClient.Recv()
+			if err == io.EOF {
+				// server closed the stream
+				log.Info("api server closed the server-side stream")
+				done <- true
+			} else if err != nil {
+				log.Error("error reading from stream: %v", err)
+				done <- true
+			}
+
+			account := resp.GetDatum().GetAccountWrapper()
+			printAccount(account, address)
+		}
+	}()
+}
+
 // printGlobalState prints the current global state
 func (r *repl) printGlobalState() {
 	resp, err := r.client.GlobalStateHash()
@@ -49,26 +112,11 @@ func (r *repl) printGlobalState() {
 func (r *repl) printAccountState() {
 	addressStr := inputNotBlank(enterAddressMsg)
 	address := gosmtypes.BytesToAddress(util.FromHex(addressStr))
-	state, err := r.client.AccountState(address)
+	account, err := r.client.AccountState(address)
 	if err != nil {
 		log.Error("failed to get account info: %v", err)
 		return
 	}
 
-	currBalance := uint64(0)
-	if state.StateCurrent.Balance != nil {
-		currBalance = state.StateCurrent.Balance.Value
-	}
-
-	projectedBalance := uint64(0)
-	if state.StateProjected.Balance != nil {
-		projectedBalance = state.StateProjected.Balance.Value
-	}
-
-	fmt.Println(printPrefix, "Address:", address.String())
-	fmt.Println(printPrefix, "Balance:", coinAmount(currBalance)) // currBalance, coinUnitName)
-	fmt.Println(printPrefix, "Nonce:", state.StateCurrent.Counter)
-	fmt.Println(printPrefix, "Projected Balance:", coinAmount(projectedBalance)) // projectedBalance, coinUnitName)
-	fmt.Println(printPrefix, "Projected Nonce:", state.StateProjected.Counter)
-	fmt.Println(printPrefix, "Projected state includes all pending transactions that haven't been added to the mesh yet.")
+	printAccount(account, address)
 }

repl/mesh_commands.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"time"
 
+	gosmtypes "github.com/spacemeshos/go-spacemesh/common/types"
+
 	"github.com/spacemeshos/CLIWallet/log"
 )
 
@@ -24,3 +26,37 @@ func (r *repl) printMeshInfo() {
 	fmt.Println(printPrefix, "Current epoch:", info.CurrentEpoch)
 	fmt.Println(printPrefix, "Genesis time:", localGenesisTime.Local().String())
 }
+
+// printCurrAccountMeshTransactions displays mesh transactions for the current account
+func (r *repl) printCurrAccountMeshTransactions() {
+	acc, err := r.getCurrent()
+	if err != nil {
+		log.Error("failed to get account", err)
+		return
+	}
+	r.printAccountMeshTransactions(acc.Address())
+}
+
+// printAccountMeshTransactions displays mesh transactions for an account
+func (r *repl) printMeshTransactions() {
+	addrStr := inputNotBlank(enterAddressMsg)
+	addr := gosmtypes.HexToAddress(addrStr)
+	r.printAccountMeshTransactions(addr)
+}
+
+// Print transaction for an account from mesh data
+func (r *repl) printAccountMeshTransactions(address gosmtypes.Address) {
+
+	// todo: request offset and total from user
+	txs, total, err := r.client.GetMeshTransactions(address, 0, 1000)
+	if err != nil {
+		log.Error("failed to print transactions: %v", err)
+		return
+	}
+
+	fmt.Println(printPrefix, fmt.Sprintf("Total mesh transactions: %d", total))
+	for _, tx := range txs {
+		printTransaction(tx)
+		fmt.Println(printPrefix, "-----")
+	}
+}

repl/prompt.go

@@ -20,13 +20,6 @@ func runPrompt(executor func(string), completer func(prompt.Document) []prompt.S
 		prompt.OptionPrefixTextColor(prompt.LightGray),
 		prompt.OptionMaxSuggestion(length),
 		prompt.OptionShowCompletionAtStart(),
-
-		// todo: fix me
-		/*
-			prompt.OptionAddKeyBind(
-				prompt.KeyBind{prompt.ControlC, func(*prompt.Buffer) {
-					_ = syscall.Kill(syscall.Getpid(), syscall.SIGINT)
-				}}),*/
 	)
 	firstTime()
 	p.Run()