pubsub: handle ctx
This commit was moved from ipfs/go-ipfs-http-client@d451a4943c5311d7e32fcf86f56bdc1eda977fc7
Łukasz Magiera committed
Feb 18, 2019 at 17:02 UTC
4a6d36d98b6527e9c13918ac901c4cbc68630b7a
1 file changed
+56
-14
client/httpapi/pubsub.go
+56
-14
@@ -57,8 +57,10 @@ func (api *PubsubAPI) Publish(ctx context.Context, topic string, message []byte)
57
}
58
59
type pubsubSub struct {
60
- io.Closer
61
- dec *json.Decoder
60
+ messages chan pubsubMessage
61
+
62
+ done chan struct{}
63
+ rcloser io.Closer
64
}
65
66
type pubsubMessage struct {
@@ -68,6 +70,7 @@ type pubsubMessage struct {
70
JTopicIDs []string `json:"topicIDs,omitempty"`
71
72
from peer.ID
73
+ err error
74
}
75
76
func (msg *pubsubMessage) From() peer.ID {
@@ -87,15 +90,20 @@ func (msg *pubsubMessage) Topics() []string {
90
}
91
92
func (s *pubsubSub) Next(ctx context.Context) (iface.PubSubMessage, error) {
90
- // TODO: handle ctx
91
-
92
- var msg pubsubMessage
93
- if err := s.dec.Decode(&msg); err != nil {
94
- return nil, err
93
+ select {
94
+ case msg, ok := <-s.messages:
95
+ if !ok {
96
+ return nil, io.EOF
97
+ }
98
+ if msg.err != nil {
99
+ return nil, msg.err
100
+ }
101
+ var err error
102
+ msg.from, err = peer.IDFromBytes(msg.JFrom)
103
+ return &msg, err
104
+ case <-ctx.Done():
105
+ return nil, ctx.Err()
106
}
96
- var err error
97
- msg.from, err = peer.IDFromBytes(msg.JFrom)
98
- return &msg, err
107
}
108
109
func (api *PubsubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (iface.PubSubSubscription, error) {
@@ -114,10 +122,44 @@ func (api *PubsubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
122
return nil, resp.Error
123
}
124
117
- return &pubsubSub{
118
- Closer: resp,
119
- dec: json.NewDecoder(resp.Output),
120
- }, nil
125
+ sub := &pubsubSub{
126
+ messages: make(chan pubsubMessage),
127
+ done: make(chan struct{}),
128
+ }
129
+
130
+ dec := json.NewDecoder(resp.Output)
131
+
132
+ go func() {
133
+ defer close(sub.messages)
134
+
135
+ for {
136
+ var msg pubsubMessage
137
+ if err := dec.Decode(&msg); err != nil {
138
+ if err == io.EOF {
139
+ return
140
+ }
141
+ msg.err = err
142
+ }
143
+
144
+ select {
145
+ case sub.messages <- msg:
146
+ case <-sub.done:
147
+ return
148
+ case <-ctx.Done():
149
+ return
150
+ }
151
+ }
152
+ }()
153
+
154
+ return sub, nil
155
+}
156
+
157
+func (s*pubsubSub) Close() error {
158
+ if s.done != nil {
159
+ close(s.done)
160
+ s.done = nil
161
+ }
162
+ return s.rcloser.Close()
163
}
164
165
func (api *PubsubAPI) core() *HttpApi {