Dummy support for progress tokens and remote destination in LocalVisitorSession
@@ -28,7 +28,7 @@ import java.util.concurrent.atomic.AtomicReference;
/**
* Local visitor session that copies and iterates through all items in the local document access.
- * Each document must be ack'ed for the session to be done visiting.
+ * Each document must be ack'ed for the session to be done visiting, unless the destinatino is remote.
* Only document puts are sent by this session, and this is done from a separate thread.
*
* @author jonmv
@@ -44,20 +44,23 @@ public class LocalVisitorSession implements VisitorSession {
private final FieldSet fieldSet;
private final AtomicReference<State> state;
private final AtomicReference<Phaser> phaser;
+ private final ProgressToken token;
public LocalVisitorSession(LocalDocumentAccess access, VisitorParameters parameters) throws ParseException {
- if (parameters.getResumeToken() != null)
- throw new UnsupportedOperationException("Continuation via progress tokens is not supported");
-
- if (parameters.getRemoteDataHandler() != null)
- throw new UnsupportedOperationException("Remote data handlers are not supported");
-
this.selector = new DocumentSelector(parameters.getDocumentSelection());
this.fieldSet = new FieldSetRepo().parse(access.getDocumentTypeManager(), parameters.fieldSet());
+ this.token = parameters.getResumeToken();
+ if (parameters.getRemoteDataHandler() == null) {
this.data = parameters.getLocalDataHandler() == null ? new VisitorDataQueue() : parameters.getLocalDataHandler();
this.data.reset();
this.data.setSession(this);
+ }
+ else {
+ if (parameters.getLocalDataHandler() != null)
+ throw new IllegalArgumentException("Cannot have both a remote and a local data handler");
+ this.data = null;
+ }
this.control = parameters.getControlHandler() == null ? new VisitorControlHandler() : parameters.getControlHandler();
this.control.reset();
@@ -98,8 +101,11 @@ public class LocalVisitorSession implements VisitorSession {
if (synchronizer != null)
synchronizer.arriveAndAwaitAdvance();
+ if (data != null)
data.onMessage(new PutDocumentMessage(new DocumentPut(copy)),
new AckToken(id));
+ else
+ outstanding.remove(id);
if (synchronizer != null)
synchronizer.arriveAndAwaitAdvance();
@@ -127,8 +133,8 @@ public class LocalVisitorSession implements VisitorSession {
}
finally {
if (synchronizer != null)
- synchronizer.arriveAndDeregister();
-
+ synchronizer.awaitAdvance(synchronizer.arriveAndDeregister());
+ if (data != null)
data.onDone();
}
}).start();
@@ -140,9 +146,10 @@ public class LocalVisitorSession implements VisitorSession {
&& control.isDone(); // Control handler has been notified
}
+ /** Returns the token set in the parameters used to create this. */
@Override
public ProgressToken getProgress() {
- throw new UnsupportedOperationException("Progress tokens are not supported");
+ return token;
}
@Override
