/
legacy_filter.go
105 lines (86 loc) · 2.6 KB
/
legacy_filter.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
package library
import (
"context"
"encoding/json"
"time"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter"
"github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter/pb"
)
type legacyFilterArgument struct {
Topic string `json:"pubsubTopic,omitempty"`
ContentFilters []*pb.FilterRequest_ContentFilter `json:"contentFilters,omitempty"`
}
func toLegacyContentFilter(filterJSON string) (legacy_filter.ContentFilter, error) {
var f legacyFilterArgument
err := json.Unmarshal([]byte(filterJSON), &f)
if err != nil {
return legacy_filter.ContentFilter{}, err
}
result := legacy_filter.ContentFilter{
Topic: f.Topic,
}
for _, cf := range f.ContentFilters {
result.ContentTopics = append(result.ContentTopics, cf.ContentTopic)
}
return result, err
}
// LegacyFilterSubscribe is used to create a subscription to a filter node to receive messages
// Deprecated: Use FilterSubscribe instead
func LegacyFilterSubscribe(filterJSON string, peerID string, ms int) error {
cf, err := toLegacyContentFilter(filterJSON)
if err != nil {
return err
}
if wakuState.node == nil {
return errWakuNodeNotReady
}
var ctx context.Context
var cancel context.CancelFunc
if ms > 0 {
ctx, cancel = context.WithTimeout(context.Background(), time.Duration(int(ms))*time.Millisecond)
defer cancel()
} else {
ctx = context.Background()
}
var fOptions []legacy_filter.FilterSubscribeOption
if peerID != "" {
p, err := peer.Decode(peerID)
if err != nil {
return err
}
fOptions = append(fOptions, legacy_filter.WithPeer(p))
} else {
fOptions = append(fOptions, legacy_filter.WithAutomaticPeerSelection())
}
_, f, err := wakuState.node.LegacyFilter().Subscribe(ctx, cf, fOptions...)
if err != nil {
return err
}
go func(f legacy_filter.Filter) {
for envelope := range f.Chan {
send("message", toSubscriptionMessage(envelope))
}
}(f)
return nil
}
// LegacyFilterUnsubscribe is used to remove a filter criteria from an active subscription with a filter node
// Deprecated: Use FilterUnsubscribe or FilterUnsubscribeAll instead
func LegacyFilterUnsubscribe(filterJSON string, ms int) error {
cf, err := toLegacyContentFilter(filterJSON)
if err != nil {
return err
}
if wakuState.node == nil {
return errWakuNodeNotReady
}
var ctx context.Context
var cancel context.CancelFunc
if ms > 0 {
ctx, cancel = context.WithTimeout(context.Background(), time.Duration(int(ms))*time.Millisecond)
defer cancel()
} else {
ctx = context.Background()
}
return wakuState.node.LegacyFilter().UnsubscribeFilter(ctx, cf)
}