Wait in destinatino until RPC layer times out request
@@ -4,9 +4,11 @@ package com.yahoo.documentapi.messagebus;
import com.yahoo.document.DocumentRemove;
import com.yahoo.documentapi.DocumentAccess;
import com.yahoo.documentapi.DocumentAccessParams;
+import com.yahoo.documentapi.ProgressToken;
import com.yahoo.documentapi.SyncParameters;
import com.yahoo.documentapi.SyncSession;
import com.yahoo.documentapi.local.LocalDocumentAccess;
+import com.yahoo.documentapi.messagebus.protocol.CreateVisitorReply;
import com.yahoo.documentapi.messagebus.protocol.DocumentMessage;
import com.yahoo.documentapi.messagebus.protocol.DocumentProtocol;
import com.yahoo.documentapi.messagebus.protocol.GetDocumentMessage;
@@ -28,6 +30,7 @@ import com.yahoo.messagebus.network.rpc.RPCNetworkParams;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.Phaser;
/**
* Mock-up destination used for testing.
@@ -36,6 +39,8 @@ import java.util.List;
*/
public class Destination implements MessageHandler {
+ final Phaser phaser = new Phaser(1);
+
private final DestinationSession session;
private final DocumentAccess access;
private final SyncSession local;
@@ -55,11 +60,9 @@ public class Destination implements MessageHandler {
session = bus.getMessageBus().createDestinationSession("session", true, this);
}
- protected void sendReply(Reply reply) {
- session.reply(reply);
- }
-
public void handleMessage(Message msg) {
+
+ phaser.arriveAndAwaitAdvance();
Reply reply = ((DocumentMessage)msg).createReply();
try {
switch (msg.getType()) {
@@ -80,6 +83,10 @@ public class Destination implements MessageHandler {
local.update(((UpdateDocumentMessage)msg).getDocumentUpdate());
break;
+ case DocumentProtocol.MESSAGE_CREATEVISITOR:
+ ((CreateVisitorReply) reply).setLastBucket(ProgressToken.FINISHED_BUCKET);
+ break;
+
default:
throw new UnsupportedOperationException("Unsupported message type '" + msg.getType() + "'.");
}
