code: func (s *Server) Subscribe(ctx context.Context, clientID string, qry query.Query, outBuffer int) (<-chan interface{}, error) {
  	s.mtx.RLock()
  	clientSubscriptions, ok := s.subscriptions[clientID]
  	if ok {
  		_, ok = clientSubscriptions[qry.String()]
  	}
  	s.mtx.RUnlock()
  	if ok {
  		return nil, ErrAlreadySubscribed
  	}
  	// We are responsible for closing this channel so we create it
  	out := make(chan interface{}, outBuffer)
  	select {
  	case s.cmds <- cmd{op: sub, clientID: clientID, query: qry, ch: out}:
  		s.mtx.Lock()
  		if _, ok = s.subscriptions[clientID]; !ok {
  			s.subscriptions[clientID] = make(map[string]query.Query)
  		}
  		// preserve original query
  		// see Unsubscribe
  		s.subscriptions[clientID][qry.String()] = qry
  		s.mtx.Unlock()
  		return out, nil
  	case <-ctx.Done():
  		return nil, ctx.Err()
  	}
  }
docs: // Subscribe creates a subscription for the given client. It accepts a channel
  // on which messages matching the given query can be received. An error will be
  // returned to the caller if the context is canceled or if subscription already
  // exist for pair clientID and query.