-
Notifications
You must be signed in to change notification settings - Fork 43
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
Showing
18 changed files
with
564 additions
and
167 deletions.
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,156 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/waku-org/go-waku/logging" | ||
| "github.com/waku-org/go-waku/waku/v2/node" | ||
| wprotocol "github.com/waku-org/go-waku/waku/v2/protocol" | ||
| "github.com/waku-org/go-waku/waku/v2/protocol/relay" | ||
| "github.com/waku-org/go-waku/waku/v2/rendezvous" | ||
| "github.com/waku-org/go-waku/waku/v2/utils" | ||
| "go.uber.org/zap" | ||
| "golang.org/x/exp/maps" | ||
| ) | ||
|
|
||
| var fwdMetaTag = []byte{102, 119, 100} //"fwd" | ||
|
|
||
| func handleRelayTopics(ctx context.Context, wg *sync.WaitGroup, wakuNode *node.WakuNode, pubSubTopicMap map[string][]string) error { | ||
| for nodeTopic, cTopics := range pubSubTopicMap { | ||
| nodeTopic := nodeTopic | ||
| _, err := wakuNode.Relay().Subscribe(ctx, wprotocol.NewContentFilter(nodeTopic, cTopics...), relay.WithoutConsumer()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if len(options.Rendezvous.Nodes) != 0 { | ||
| // Register the node in rendezvous point | ||
| iter := rendezvous.NewRendezvousPointIterator(options.Rendezvous.Nodes) | ||
|
|
||
| wg.Add(1) | ||
| go func(nodeTopic string) { | ||
| t := time.NewTicker(rendezvous.RegisterDefaultTTL) | ||
| defer t.Stop() | ||
| defer wg.Done() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-t.C: | ||
| // Register in rendezvous points periodically | ||
| wakuNode.Rendezvous().RegisterWithNamespace(ctx, nodeTopic, iter.RendezvousPoints()) | ||
| } | ||
| } | ||
| }(nodeTopic) | ||
|
|
||
| wg.Add(1) | ||
| go func(nodeTopic string) { | ||
| defer wg.Done() | ||
| desiredOutDegree := wakuNode.Relay().Params().D | ||
| t := time.NewTicker(7 * time.Second) | ||
| defer t.Stop() | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-t.C: | ||
| peerCnt := len(wakuNode.Relay().PubSub().ListPeers(nodeTopic)) | ||
| peersToFind := desiredOutDegree - peerCnt | ||
| if peersToFind <= 0 { | ||
| continue | ||
| } | ||
|
|
||
| rp := <-iter.Next(ctx) | ||
| if rp == nil { | ||
| continue | ||
| } | ||
| ctx, cancel := context.WithTimeout(ctx, 7*time.Second) | ||
| wakuNode.Rendezvous().DiscoverWithNamespace(ctx, nodeTopic, rp, peersToFind) | ||
| cancel() | ||
| } | ||
| } | ||
| }(nodeTopic) | ||
|
|
||
| } | ||
| } | ||
|
|
||
| // Protected topics | ||
| for _, protectedTopic := range options.Relay.ProtectedTopics { | ||
| if err := wakuNode.Relay().AddSignedTopicValidator(protectedTopic.Topic, protectedTopic.PublicKey); err != nil { | ||
| return nonRecoverErrorMsg("could not add signed topic validator: %w", err) | ||
| } | ||
| } | ||
|
|
||
| err := bridgeTopics(ctx, wg, wakuNode) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func bridgeTopics(ctx context.Context, wg *sync.WaitGroup, wakuNode *node.WakuNode) error { | ||
| // Bridge topics | ||
| bridgedTopics := make(map[string]map[string]struct{}) | ||
| bridgedTopicsSet := make(map[string]struct{}) | ||
| for _, topics := range options.Relay.BridgeTopics { | ||
| _, ok := bridgedTopics[topics.FromTopic] | ||
| if !ok { | ||
| bridgedTopics[topics.FromTopic] = make(map[string]struct{}) | ||
| } | ||
|
|
||
| bridgedTopics[topics.FromTopic][topics.ToTopic] = struct{}{} | ||
| bridgedTopicsSet[topics.FromTopic] = struct{}{} | ||
| bridgedTopicsSet[topics.ToTopic] = struct{}{} | ||
| } | ||
|
|
||
| // Make sure all topics are subscribed | ||
| for _, topic := range maps.Keys(bridgedTopicsSet) { | ||
| if !wakuNode.Relay().IsSubscribed(topic) { | ||
| _, err := wakuNode.Relay().Subscribe(ctx, wprotocol.NewContentFilter(topic), relay.WithoutConsumer()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
|
|
||
| for fromTopic, toTopics := range bridgedTopics { | ||
| subscriptions, err := wakuNode.Relay().Subscribe(ctx, wprotocol.NewContentFilter(fromTopic)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| topics := maps.Keys(toTopics) | ||
| for _, subscription := range subscriptions { | ||
| wg.Add(1) | ||
| go func(subscription *relay.Subscription, topics []string) { | ||
| defer wg.Done() | ||
| for env := range subscription.Ch { | ||
| for _, topic := range topics { | ||
| // HACK: message has been already fwded | ||
| metaLen := len(env.Message().Meta) | ||
| fwdTagLen := len(fwdMetaTag) | ||
| if metaLen >= fwdTagLen && bytes.Equal(env.Message().Meta[metaLen-fwdTagLen:], fwdMetaTag) { | ||
| continue | ||
| } | ||
|
|
||
| // HACK: We append magic numbers here, just so the pubsub message ID will change | ||
| env.Message().Meta = append(env.Message().Meta, fwdMetaTag...) | ||
| _, err := wakuNode.Relay().Publish(ctx, env.Message(), relay.WithPubSubTopic(topic)) | ||
| if err != nil { | ||
| utils.Logger().Warn("could not bridge message", logging.HexString("hash", env.Hash()), | ||
| zap.String("fromTopic", env.PubsubTopic()), zap.String("toTopic", topic), | ||
| zap.String("contentTopic", env.Message().ContentTopic), zap.Error(err)) | ||
| } | ||
| } | ||
| } | ||
| }(subscription, topics) | ||
| } | ||
| } | ||
|
|
||
| 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package cliutils | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "golang.org/x/exp/slices" | ||
| ) | ||
|
|
||
| type BridgeTopic struct { | ||
| FromTopic string | ||
| ToTopic string | ||
| } | ||
|
|
||
| func (p BridgeTopic) String() string { | ||
| return fmt.Sprintf("%s:%s", p.FromTopic, p.ToTopic) | ||
| } | ||
|
|
||
| type BridgeTopicSlice struct { | ||
| Values *[]BridgeTopic | ||
| } | ||
|
|
||
| func (k *BridgeTopicSlice) Set(value string) error { | ||
| topicParts := strings.Split(value, ":") | ||
| if len(topicParts) != 2 { | ||
| return errors.New("expected from_topic:to_topic") | ||
| } | ||
|
|
||
| for i := range topicParts { | ||
| topicParts[i] = strings.TrimSpace(topicParts[i]) | ||
| } | ||
|
|
||
| if slices.Contains(topicParts, "") { | ||
| return errors.New("topic can't be empty") | ||
| } | ||
|
|
||
| *k.Values = append(*k.Values, BridgeTopic{ | ||
| FromTopic: topicParts[0], | ||
| ToTopic: topicParts[1], | ||
| }) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (k *BridgeTopicSlice) String() string { | ||
| if k.Values == nil { | ||
| return "" | ||
| } | ||
| var output []string | ||
| for _, v := range *k.Values { | ||
| output = append(output, v.String()) | ||
| } | ||
|
|
||
| return strings.Join(output, ", ") | ||
| } |
Oops, something went wrong.