final DocumentBridge bridge) {
String topic = Constants.Topic.STORE + "/" + id + Constants.Topic.WATCH;
if (bus instanceof ReliableSubscribeBus) {
((ReliableSubscribeBus) bus).synchronizeSequenceNumber(topic, version - 1);
}
final Registration handlerReg =
bus.subscribe(topic, new Handler<Message<JsonObject>>() {
Transformer<CollaborativeOperation> transformer = new CollaborativeTransformer();
@Override
public void handle(Message<JsonObject> message) {
JsonObject body = message.body();
CollaborativeOperation op = transformer.createOperation(body);
bridge.consume(op);
}
});
bridge.setOutputSink(new OutputSink() {
@Override
public void close() {
handlerReg.unregister();
}
@Override
public void consume(CollaborativeOperation op) {
}