feat: servicer module implementation #19
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
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,23 @@ | ||
| syntax = "proto3"; | ||
| package poktroll.servicer; | ||
|
|
||
| option go_package = "poktroll/x/servicer/types"; | ||
|
|
||
| message Relay { | ||
| RelayRequest req = 1; | ||
| RelayResponse res = 2; | ||
| } | ||
|
|
||
| message RelayRequest { | ||
| string method = 1; | ||
| string url = 2; | ||
| map<string, string> headers = 3; | ||
| bytes payload = 4; | ||
| } | ||
|
|
||
| message RelayResponse { | ||
| bytes payload = 1; | ||
| int32 status_code = 2; | ||
| map<string, string> headers = 3; | ||
| bytes signature = 4; | ||
| } |
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,22 @@ | ||
| syntax = "proto3"; | ||
|
There was a problem hiding this comment. Remove this and import from the session module |
||
| package poktroll.servicer; | ||
|
|
||
| option go_package = "poktroll/x/servicer/types"; | ||
|
|
||
| message Session { | ||
| string id = 1; // a universally unique ID for the session | ||
| int64 session_number = 2; // a monotonically increasing number representing the # on the chain | ||
| int64 session_height = 3; // the height at which the session starts | ||
| bytes block_hash = 4; // the hash of the block at which the session starts | ||
| int64 num_session_blocks = 5; // the number of blocks the session is valid from | ||
| // CONSIDERATION: Should we add a `RelayChain` enum and use it across the board? | ||
| // CONSIDERATION: Should a single session support multiple relay chains? | ||
| // TECHDEBT: Do we need backwards with v0? https://docs.pokt.network/supported-blockchains/ | ||
| string relay_chain = 6; // the relay chain the session is valid for | ||
| // CONSIDERATION: Should a single session support multiple geo zones? | ||
| string geo_zone = 7; // the target geographic region where the actors are present | ||
| // core.Actor application = 7; // the application that is being served | ||
| // IMPROVE: `map<string, core.Actor>` with the address as the key can simplify and optimize the logic on the clients | ||
| // repeated core.Actor servicers = 8; // the set of servicers that are serving the application | ||
| // repeated core.Actor fishermen = 9; // the set of fishermen that are fishing for servicers | ||
| } | ||
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,101 @@ | ||
| package utils | ||
|
|
||
| import "sync" | ||
|
|
||
| // Observable is a generic interface that allows multiple subscribers to read from a single channel | ||
| type Observable[V any] interface { | ||
| Subscribe() Subscription[V] | ||
| } | ||
|
|
||
| type ObservableImpl[V any] struct { | ||
| mu sync.RWMutex | ||
| ch <-chan V // private channel that is used to emit values to subscribers | ||
| subscribers []chan V // subscribers is a list of channels that are subscribed to the observable | ||
| closed bool | ||
| } | ||
|
|
||
| // Creates a new observable which emissions are controlled by the emitter channel | ||
| func NewControlledObservable[V any](emitter chan V) (Observable[V], chan V) { | ||
| // If the caller does not provide an emitter, create a new one and return it | ||
| e := make(chan V) | ||
| if emitter != nil { | ||
| e = emitter | ||
| } | ||
| o := &ObservableImpl[V]{sync.RWMutex{}, e, []chan V{}, false} | ||
|
|
||
| // Start listening to the emitter and emit values to subscribers | ||
| go o.listen(emitter) | ||
|
|
||
| return o, emitter | ||
| } | ||
|
|
||
| // Get a subscription to the observable | ||
| func (o *ObservableImpl[V]) Subscribe() Subscription[V] { | ||
| o.mu.Lock() | ||
| defer o.mu.Unlock() | ||
|
|
||
| // Create a channel for the subscriber and append it to the subscribers list | ||
| ch := make(chan V, 1) | ||
| o.subscribers = append(o.subscribers, ch) | ||
|
|
||
| // Removal function used when unsubscribing from the observable | ||
| removeFromObservable := func() { | ||
| o.mu.Lock() | ||
| defer o.mu.Unlock() | ||
|
|
||
| for i, s := range o.subscribers { | ||
| if ch == s { | ||
| o.subscribers = append(o.subscribers[:i], o.subscribers[i+1:]...) | ||
| break | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Subscription gets its closed state from the observable | ||
| return &SubscriptionImpl[V]{ch, o.closed, removeFromObservable} | ||
| } | ||
|
|
||
| // Listen to the emitter and emit values to subscribers | ||
| // This function is blocking and should be run in a goroutine | ||
| func (o *ObservableImpl[V]) listen(emitter <-chan V) { | ||
| for v := range emitter { | ||
| // Lock for o.subscribers slice as it can be modified by subscribers | ||
| o.mu.RLock() | ||
| for _, ch := range o.subscribers { | ||
| ch <- v | ||
| } | ||
| o.mu.RUnlock() | ||
| } | ||
|
|
||
| // Here we know that the emitter has been closed, all subscribers should be closed as well | ||
| o.mu.Lock() | ||
| o.closed = true | ||
| for _, ch := range o.subscribers { | ||
| close(ch) | ||
| o.subscribers = []chan V{} | ||
| } | ||
| o.mu.Lock() | ||
| } | ||
|
|
||
| // Subscription is a generic interface that provide access to the underlying channel | ||
| // and allows unsubscribing from an observable | ||
| type Subscription[V any] interface { | ||
| Unsubscribe() | ||
| Ch() <-chan V | ||
| } | ||
|
|
||
| type SubscriptionImpl[V any] struct { | ||
| ch chan V | ||
| closed bool | ||
| removeFromObservable func() | ||
| } | ||
|
|
||
| func (s *SubscriptionImpl[V]) Unsubscribe() { | ||
| close(s.ch) | ||
| s.closed = true | ||
| s.removeFromObservable() | ||
| } | ||
|
|
||
| func (s *SubscriptionImpl[V]) Ch() <-chan V { | ||
| return s.ch | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In the future, I was thinking each service can act as multiple actors. I think it would also simplify the logic by initializing all of them (none of the if else statements) and it would just be a noop inside the actual module if the flagis off.