pair_programming_code

ten-protocol · Oct 20, 2023 · e877138 · e877138
1 parent 206f45a
commit e877138
Show file tree

Hide file tree

Showing 3 changed files with 218 additions and 131 deletions.
diff --git a/integration/obscurogateway/obscurogateway_test.go b/integration/obscurogateway/obscurogateway_test.go
@@ -345,6 +345,7 @@ func TestObscuroGatewaySubscriptionsWithMultipleAccounts(t *testing.T) {
 	fmt.Printf("Tx1 was included in block %d\n", intTx1Receipt.BlockNumber)
 	fmt.Printf("Tx2 was included in block %d\n", intTx2Receipt.BlockNumber)
 
+	time.Sleep(30 * time.Second)
 	fmt.Println("user0 received logs: ", len(user0logs))
 	fmt.Println("user1 received logs: ", len(user1logs))
 	fmt.Println("user2 received logs: ", len(user2logs))
@@ -360,6 +361,7 @@ func TestObscuroGatewaySubscriptionsWithMultipleAccounts(t *testing.T) {
 	// FIXME: Currently we receive only 2 events, because only the first account on each client actually subscribe
 	// assert.Equal(t, 3, len(user2logs))
 
+	time.Sleep(time.Hour)
 	// Gracefully shutdown
 	err = obscuroGwContainer.Stop()
 	assert.NoError(t, err)
@@ -619,9 +621,10 @@ func subscribeToEvents(addresses []gethcommon.Address, topics [][]gethcommon.Has
 			case err := <-subscription.Err():
 				fmt.Printf("Error from logs subscription: %v\n", err)
 				return
-			case log := <-logsCh:
+			case log2 := <-logsCh:
+				fmt.Println(log2)
 				// append logs to be visible from the main thread
-				*logs = append(*logs, log)
+				*logs = append(*logs, log2)
 			}
 		}
 	}()

diff --git a/tools/walletextension/accountmanager/account_manager.go b/tools/walletextension/accountmanager/account_manager.go
@@ -1,12 +1,14 @@
 package accountmanager
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
-
+	"github.com/obscuronet/go-obscuro/go/common/log"
 	"github.com/obscuronet/go-obscuro/tools/walletextension/subscriptions"
+	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/eth/filters"
 
@@ -70,7 +72,7 @@ func (m *AccountManager) ProxyRequest(rpcReq *wecommon.RPCRequest, rpcResp *inte
 		}
 		return err
 
-		//// fetch the clients from a topic (todo: @ziga - delete it)
+		// fetch the clients from a topic (todo: @ziga - delete it)
 		//for _, client := range clients {
 		//	return m.executeSubscribe(client, rpcReq, rpcResp, userConn)
 		//}
@@ -279,64 +281,64 @@ func searchDataFieldForAccount(callParams map[string]interface{}, accClients map
 	return nil, fmt.Errorf("no known account found in data bytes")
 }
 
-//func (m *AccountManager) executeSubscribe(client rpc.Client, req *wecommon.RPCRequest, resp *interface{}, userConn userconn.UserConn) error { //nolint: gocognit
-//	if len(req.Params) == 0 {
-//		return fmt.Errorf("could not subscribe as no subscription namespace was provided")
-//	}
-//	m.logger.Info(fmt.Sprintf("Subscribing client: %s for request: %s", client, req))
-//	ch := make(chan common.IDAndLog)
-//	subscription, err := client.Subscribe(context.Background(), resp, rpc.SubscribeNamespace, ch, req.Params...)
-//	if err != nil {
-//		return fmt.Errorf("could not call %s with params %v. Cause: %w", req.Method, req.Params, err)
-//	}
-//
-//	// We listen for incoming messages on the subscription.
-//	go func() {
-//		for {
-//			select {
-//			case idAndLog := <-ch:
-//				if userConn.IsClosed() {
-//					m.logger.Info("received log but websocket was closed on subscription", log.SubIDKey, idAndLog.SubID)
-//					return
-//				}
-//
-//				jsonResponse, err := wecommon.PrepareLogResponse(idAndLog)
-//				if err != nil {
-//					m.logger.Error("could not marshal log response to JSON on subscription.", log.SubIDKey, idAndLog.SubID, log.ErrKey, err)
-//					continue
-//				}
-//
-//				m.logger.Trace(fmt.Sprintf("Forwarding log from Obscuro node: %s", jsonResponse), log.SubIDKey, idAndLog.SubID)
-//				err = userConn.WriteResponse(jsonResponse)
-//				if err != nil {
-//					m.logger.Error("could not write the JSON log to the websocket on subscription %", log.SubIDKey, idAndLog.SubID, log.ErrKey, err)
-//					continue
-//				}
-//
-//			case err = <-subscription.Err():
-//				// An error on this channel means the subscription has ended, so we exit the loop.
-//				if userConn != nil && err != nil {
-//					userConn.HandleError(err.Error())
-//				}
-//
-//				return
-//			}
-//		}
-//	}()
-//
-//	// We periodically check if the websocket is closed, and terminate the subscription.
-//	go func() {
-//		for {
-//			if userConn.IsClosed() {
-//				subscription.Unsubscribe()
-//				return
-//			}
-//			time.Sleep(100 * time.Millisecond)
-//		}
-//	}()
-//
-//	return nil
-//}
+func (m *AccountManager) executeSubscribe(client rpc.Client, req *wecommon.RPCRequest, resp *interface{}, userConn userconn.UserConn) error { //nolint: gocognit
+	if len(req.Params) == 0 {
+		return fmt.Errorf("could not subscribe as no subscription namespace was provided")
+	}
+	m.logger.Info(fmt.Sprintf("Subscribing client: %s for request: %s", client, req))
+	ch := make(chan common.IDAndLog)
+	subscription, err := client.Subscribe(context.Background(), resp, rpc.SubscribeNamespace, ch, req.Params...)
+	if err != nil {
+		return fmt.Errorf("could not call %s with params %v. Cause: %w", req.Method, req.Params, err)
+	}
+
+	// We listen for incoming messages on the subscription.
+	go func() {
+		for {
+			select {
+			case idAndLog := <-ch:
+				if userConn.IsClosed() {
+					m.logger.Info("received log but websocket was closed on subscription", log.SubIDKey, idAndLog.SubID)
+					return
+				}
+
+				jsonResponse, err := wecommon.PrepareLogResponse(idAndLog)
+				if err != nil {
+					m.logger.Error("could not marshal log response to JSON on subscription.", log.SubIDKey, idAndLog.SubID, log.ErrKey, err)
+					continue
+				}
+
+				m.logger.Trace(fmt.Sprintf("Forwarding log from Obscuro node: %s", jsonResponse), log.SubIDKey, idAndLog.SubID)
+				err = userConn.WriteResponse(jsonResponse)
+				if err != nil {
+					m.logger.Error("could not write the JSON log to the websocket on subscription %", log.SubIDKey, idAndLog.SubID, log.ErrKey, err)
+					continue
+				}
+
+			case err = <-subscription.Err():
+				// An error on this channel means the subscription has ended, so we exit the loop.
+				if userConn != nil && err != nil {
+					userConn.HandleError(err.Error())
+				}
+
+				return
+			}
+		}
+	}()
+
+	// We periodically check if the websocket is closed, and terminate the subscription.
+	go func() {
+		for {
+			if userConn.IsClosed() {
+				subscription.Unsubscribe()
+				return
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+	}()
+
+	return nil
+}
 
 func submitCall(client *rpc.EncRPCClient, req *wecommon.RPCRequest, resp *interface{}) error {
 	if req.Method == rpc.Call || req.Method == rpc.EstimateGas {