"database/sql"
"encoding/json"
"errors"
+ "reflect"
"sync"
"sync/atomic"
"time"
sess.mtx.Unlock()
sub.sendOldEvents(sess)
return nil
+ } else if sub.Method == "unsubscribe" {
+ sess.mtx.Lock()
+ found := false
+ for i, s := range sess.subscriptions {
+ if !reflect.DeepEqual(s.Filters, sub.Filters) {
+ continue
+ }
+ copy(sess.subscriptions[i:], sess.subscriptions[i+1:])
+ sess.subscriptions = sess.subscriptions[:len(sess.subscriptions)-1]
+ found = true
+ break
+ }
+ sess.mtx.Unlock()
+ sess.log.WithField("sub", sub).WithField("found", found).Debug("unsubscribe")
+ if found {
+ sess.sendq <- v0subscribeOK
+ return nil
+ }
} else {
sess.log.WithField("Method", sub.Method).Info("unknown method")
}
conn, r, w := s.testClient()
defer conn.Close()
- c.Check(w.Encode(map[string]interface{}{
- "method": "subscribe",
- "filters": [][]interface{}{{"event_type", "in", []string{"update"}}},
- }), check.IsNil)
- s.expectStatus(c, r, 200)
+ cmd := func(method, eventType string, status int) {
+ c.Check(w.Encode(map[string]interface{}{
+ "method": method,
+ "filters": [][]interface{}{{"event_type", "in", []string{eventType}}},
+ }), check.IsNil)
+ s.expectStatus(c, r, status)
+ }
+ cmd("subscribe", "update", 200)
+ cmd("subscribe", "update", 200)
+ cmd("subscribe", "create", 200)
+ cmd("subscribe", "update", 200)
+ cmd("unsubscribe", "blip", 400)
+ cmd("unsubscribe", "create", 200)
+ cmd("unsubscribe", "update", 200)
go s.emitEvents(nil)
lg := s.expectLog(c, r)
c.Check(lg.EventType, check.Equals, "update")
+
+ cmd("unsubscribe", "update", 200)
+ cmd("unsubscribe", "update", 200)
+ cmd("unsubscribe", "update", 400)
}
func (s *v0Suite) TestLastLogID(c *check.C) {