code: private void realSubscribe(Subscriber subs) {
          AtomicInteger doneCount = new AtomicInteger(shards.size());
          ChangeReceiver receiver = subs.getReceiver();
          ArrayList<Subscriber> subsList = new ArrayList();
          shards.forEach( shard -> {
              Subscriber shardSubs = new Subscriber(subs.getFilter(), change -> {
                  if (change.getType() == ChangeMessage.QUERYDONE) {
                      int count = doneCount.decrementAndGet();
                      if (count == 0) {
                          receiver.receive(change);
                      }
                  } else {
                      receiver.receive(change);
                  }
              });
              subsList.add(shardSubs);
              shard.subscribe(shardSubs);
          });
          subsMap.put(subs,subsList);
      }
docs: actually subscribes at datanodes