anchor.diff:    }
   
     void onUnsubscribe(Client c, String topicUri) {
  +    final uri = curie.decode(topicUri);
  +    c.topics.remove(uri);
  +
  +    _unsubscribe(c, topicUri);
  +  }
  +
  +  void onPublish(Client c, String topicUri, event, [exclude, eligible]) {
  +    // TODO: handle exclude, eligible.
  +    publish(topicUri, event);
  +  }
  +
  +  void _unsubscribe(Client c, String topicUri) {
       var uri = curie.decode(topicUri);
  -
  -    c.topics.remove(uri);
   
       if (topicMap.containsKey(uri)) {
         topicMap[uri].remove(c);
  
positive.code:     }
    }
  
    void onPublish(Client c, String topicUri, event, [exclude, eligible]) {
      // TODO: handle exclude, eligible.
      publish(topicUri, event);
    }
  
    /**
     * Publish an event to all the subscribed clients.
     */
    void publish(String topicUri, event) {
      final uri = curie.decode(topicUri),
            subscribers = topicMap[uri],
            garbage = [];
  
      subscribers.forEach((client) {
        if (clients.contains(client)) {
          client.event(topicUri, event);
        } else {
          // The client is disconnected, so gc.
          garbage.add(client);
        }
      });
  
      // TODO: perfome gc when the client is disconnected?
      if (!garbage.isEmpty) {
        garbage.forEach((c) => subscribers.remove(c));
      }
    }
  }
negative.code:     clients.add(c);
  
      conn.onClosed = (int status, String reason) {
        c.topics.forEach((t) => _unsubscribe(c, t));
        clients.remove(c);
      };
  
  