| 170 | } |
| 171 | |
| 172 | func (s *Server) subscribe(ctx context.Context, clientID string, query Query, outCapacity int) (*Subscription, error) { |
| 173 | s.mtx.RLock() |
| 174 | clientSubscriptions, ok := s.subscriptions[clientID] |
| 175 | if ok { |
| 176 | _, ok = clientSubscriptions[query.String()] |
| 177 | } |
| 178 | s.mtx.RUnlock() |
| 179 | if ok { |
| 180 | return nil, ErrAlreadySubscribed |
| 181 | } |
| 182 | |
| 183 | subscription := NewSubscription(outCapacity) |
| 184 | select { |
| 185 | case s.cmds <- cmd{op: sub, clientID: clientID, query: query, subscription: subscription}: |
| 186 | s.mtx.Lock() |
| 187 | if _, ok = s.subscriptions[clientID]; !ok { |
| 188 | s.subscriptions[clientID] = make(map[string]struct{}) |
| 189 | } |
| 190 | s.subscriptions[clientID][query.String()] = struct{}{} |
| 191 | s.mtx.Unlock() |
| 192 | return subscription, nil |
| 193 | case <-ctx.Done(): |
| 194 | return nil, ctx.Err() |
| 195 | case <-s.Quit(): |
| 196 | return nil, errors.New("service is shutting down") |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | // Unsubscribe removes the subscription on the given query. An error will be |
| 201 | // returned to the caller if the context is canceled or if subscription does |