Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package subscriptions | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| gethlog "github.com/ethereum/go-ethereum/log" | ||
| "github.com/obscuronet/go-obscuro/go/common" | ||
| "github.com/obscuronet/go-obscuro/go/common/log" | ||
| "github.com/obscuronet/go-obscuro/go/rpc" | ||
| wecommon "github.com/obscuronet/go-obscuro/tools/walletextension/common" | ||
| "github.com/obscuronet/go-obscuro/tools/walletextension/userconn" | ||
| ) | ||
|
|
||
| type SubscriptionManager struct { | ||
| logger gethlog.Logger | ||
| } | ||
|
|
||
| func New(logger gethlog.Logger) *SubscriptionManager { | ||
| return &SubscriptionManager{ | ||
| logger: logger, | ||
| } | ||
| } | ||
|
|
||
| func (s *SubscriptionManager) HandleNewSubscriptions(clients []rpc.Client, req *wecommon.RPCRequest, resp *interface{}, userConn userconn.UserConn) error { | ||
| if len(req.Params) == 0 { | ||
| return fmt.Errorf("could not subscribe as no subscription namespace was provided") | ||
| } | ||
|
|
||
| if !userConn.SupportsSubscriptions() { | ||
| return errors.New("userConn doesn't support subscriptions") | ||
| } | ||
|
|
||
| // create a chanel that will collect the data from all subscriptions | ||
| commonChannel := make(chan common.IDAndLog) | ||
|
|
||
| // TODO: Create a buffered channel and perform deduplication of logs or implement additional logic to filter logs | ||
|
|
||
| // TODO: Do periodic checks if userConn is closed and unsubscribe from all subscriptions | ||
|
|
||
| // Send all logs from common channel to user (via userConn) | ||
| go func() { | ||
| for { | ||
| select { | ||
| case idAndLog := <-commonChannel: | ||
| if userConn.IsClosed() { | ||
| s.logger.Info("received log but websocket was closed on subscription", log.SubIDKey, idAndLog.SubID) | ||
| return | ||
| } | ||
|
|
||
| jsonResponse, err := wecommon.PrepareLogResponse(idAndLog) | ||
| if err != nil { | ||
| s.logger.Error("could not marshal log response to JSON on subscription.", log.SubIDKey, idAndLog.SubID, log.ErrKey, err) | ||
| continue | ||
| } | ||
|
|
||
| err = userConn.WriteResponse(jsonResponse) | ||
| if err != nil { | ||
| s.logger.Error("could not write the JSON log to the websocket on subscription %", log.SubIDKey, idAndLog.SubID, log.ErrKey, err) | ||
| continue | ||
| } | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| // loop over all clients and create a new subscription for each of them | ||
| s.logger.Info(fmt.Sprintf("Subscribing to: %d clients", len(clients))) | ||
| for _, client := range clients { | ||
| s.logger.Info(fmt.Sprintf("Subscribing for an event with client: %s", client)) | ||
| subscriptionID, err := s.addSubscription(client, req, commonChannel) | ||
| if err != nil { | ||
| s.logger.Info(fmt.Sprintf("Error: %v", err)) | ||
| } | ||
| s.logger.Info(fmt.Sprintf("Subscribed with subscription ID: %s", subscriptionID)) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *SubscriptionManager) addSubscription(client rpc.Client, req *wecommon.RPCRequest, commonChannel chan common.IDAndLog) (string, error) { | ||
| s.logger.Info(fmt.Sprintf("Subscribing client: %s for request: %s", client, req)) | ||
|
|
||
| // TODO: Create a subscription with Obscuro Node (with new web socket connection each time!) | ||
|
|
||
| // TODO: | ||
| // Do something similar as in HandleNewSubscriptions | ||
| //and check for events comming to this channel and forward them to a common channel | ||
|
|
||
| return "", nil | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters