diff --git a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java index 59dd6851..2b2521cb 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java @@ -26,6 +26,8 @@ package org.forgerock.openicf.common.rpc; +import java.util.concurrent.atomic.AtomicBoolean; + import org.forgerock.util.promise.ExceptionHandler; import org.forgerock.util.promise.ResultHandler; @@ -43,10 +45,34 @@ public abstract class LocalRequest> localRequests = new ConcurrentSkipListMap>(); + /** + * Cancel messages that arrived before the addressed operation registered + * itself in {@link #localRequests} are parked here (request id to expiry + * timestamp) and applied when {@link #receiveRequest} registers the + * request. Request ids are never reused within a group, so a parked + * cancel can only ever match the operation it was sent for; entries whose + * operation never arrives are dropped after {@link #PENDING_CANCEL_TTL_MS}. + * The park in {@link #receiveRequestCancel} is unconditional, so a cancel + * that loses the race with completion (the operation already produced its + * result and unregistered) also leaves an entry behind for the full TTL; + * such entries are harmless and are removed by the next purge. + */ + private final ConcurrentMap pendingCancels = + new ConcurrentHashMap(); + + private static final long PENDING_CANCEL_TTL_MS = 60L * 1000L; + protected final CopyOnWriteArrayList> webSockets = new CopyOnWriteArrayList>(); @@ -192,14 +212,40 @@ public , V, E extends Exception> R trySub return remoteRequest; } + /** + * Registers the given request so responses and cancel messages can be + * dispatched to it. + * + * @return the registered request, or {@code null} when a cancel for its + * request id arrived before registration - the request is + * cancelled and must not be executed. + */ public > R receiveRequest( final R localRequest) { + purgeExpiredPendingCancels(); LocalRequest tmp = localRequests.putIfAbsent(localRequest.getRequestId(), localRequest); if (null != tmp && !tmp.equals(localRequest)) { throw new IllegalStateException("Request has been registered with id: " + localRequest.getRequestId()); } + // Registration and receiveRequestCancel write the two maps in + // opposite order, so whichever call runs second is guaranteed to see + // the other's entry and deliver the cancel. The verdict must come + // from the request's own state rather than from winning the park + // removal: receiveRequestCancel may find the request in localRequests + // right after the putIfAbsent above, consume its own park and cancel + // the request directly - this thread then observes no parked entry. + if (null != pendingCancels.remove(localRequest.getRequestId())) { + localRequest.cancel(); + } + if (localRequest.isCancelled()) { + // cancel() removes the registration, but only when the cancelling + // side found it - make sure a request reported dead is never left + // registered. + localRequests.remove(localRequest.getRequestId(), localRequest); + return null; + } return localRequest; } @@ -216,13 +262,31 @@ public > R receive } public LocalRequest receiveRequestCancel(long messageId) { + purgeExpiredPendingCancels(); + // Park first, then look up: the operation may be racing through + // registration on another thread, and this order guarantees that at + // least one side observes the other (see receiveRequest). + pendingCancels.put(messageId, System.currentTimeMillis() + PENDING_CANCEL_TTL_MS); LocalRequest tmp = localRequests.remove(messageId); if (null != tmp) { + pendingCancels.remove(messageId); tmp.cancel(); } return tmp; } + private void purgeExpiredPendingCancels() { + if (pendingCancels.isEmpty()) { + return; + } + final long now = System.currentTimeMillis(); + for (Map.Entry entry : pendingCancels.entrySet()) { + if (entry.getValue() < now) { + pendingCancels.remove(entry.getKey(), entry.getValue()); + } + } + } + // -- Pair of methods to Cancel pending Request End -- // -- Pair of methods to Communicate pending Request Start -- diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java new file mode 100644 index 00000000..e477782b --- /dev/null +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java @@ -0,0 +1,211 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.forgerock.openicf.common.rpc; + +import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; + +import org.forgerock.openicf.common.rpc.impl.TestConnectionContext; +import org.forgerock.openicf.common.rpc.impl.TestConnectionGroup; +import org.forgerock.openicf.common.rpc.impl.TestLocalRequest; +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * Tests the explicit {@link LocalRequest#register()} step and the handling of + * a {@code CancelOpRequest} that is processed before the addressed operation + * has registered itself (OpenIdentityPlatform/OpenICF#112). + */ +public class LocalRequestRegistrationTest, H, TestConnectionContext>> { + + /** + * Minimal connection holder: the tests never transmit anything, they only + * need {@link #getRemoteConnectionContext()} to reach the group. + */ + private class StubConnectionHolder implements + RemoteConnectionHolder, H, TestConnectionContext> { + + private final TestConnectionGroup group; + + private StubConnectionHolder(TestConnectionGroup group) { + this.group = group; + } + + public TestConnectionContext getRemoteConnectionContext() { + return group.getRemoteConnectionContext(); + } + + public Future sendBytes(byte[] data) { + throw new UnsupportedOperationException(); + } + + public Future sendString(String data) { + throw new UnsupportedOperationException(); + } + + public void sendPing(byte[] applicationData) throws Exception { + throw new UnsupportedOperationException(); + } + + public void sendPong(byte[] applicationData) throws Exception { + throw new UnsupportedOperationException(); + } + + public void close() { + } + } + + @SuppressWarnings("unchecked") + private H newSocket(TestConnectionGroup group) { + return (H) new StubConnectionHolder(group); + } + + @Test + public void testConstructorDoesNotRegister() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + new TestLocalRequest(1L, newSocket(group)); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testRegisterThenCancel() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + + Assert.assertTrue(request.register()); + Assert.assertTrue(group.getLocalRequests().contains(1L)); + + Assert.assertSame(group.receiveRequestCancel(1L), request); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testCancelBeforeRegistrationIsParked() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + + // The cancel is processed before the operation registered itself. + Assert.assertNull(group.receiveRequestCancel(1L)); + + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + Assert.assertFalse(request.register()); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testParkedCancelDoesNotAffectOtherRequests() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + + Assert.assertNull(group.receiveRequestCancel(1L)); + + TestLocalRequest other = new TestLocalRequest(2L, newSocket(group)); + Assert.assertTrue(other.register()); + Assert.assertFalse(other.isCancelled()); + Assert.assertTrue(group.getLocalRequests().contains(2L)); + } + + @Test + public void testCancelIsDeliveredOnce() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + final int[] cancelCount = new int[1]; + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)) { + public boolean tryCancel() { + cancelCount[0]++; + return super.tryCancel(); + } + }; + + Assert.assertTrue(request.register()); + Assert.assertTrue(request.cancel()); + Assert.assertFalse(request.cancel()); + Assert.assertEquals(cancelCount[0], 1); + } + + /** + * Replays the interleaving where the whole {@code receiveRequestCancel} + * runs between the registration's {@code putIfAbsent} and its pending + * cancel check: the re-registration below stands in for the registering + * thread resuming after the cancel was delivered directly. The verdict + * must come from the request's own cancelled state, so register() must + * still report the request dead. + */ + @Test + public void testRegisterObservesDirectlyDeliveredCancel() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + + Assert.assertTrue(request.register()); + Assert.assertSame(group.receiveRequestCancel(1L), request); + + Assert.assertFalse(request.register()); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + /** + * Races {@link LocalRequest#register()} against + * {@link TestConnectionGroup#receiveRequestCancel(long)}. Whatever the + * interleaving, the cancel must be delivered exactly once and the request + * must end up unregistered. + */ + @Test + public void testConcurrentRegisterAndCancel() throws Exception { + final TestConnectionGroup group = new TestConnectionGroup("test"); + final ExecutorService executor = Executors.newFixedThreadPool(2); + try { + for (long requestId = 1; requestId <= 1000; requestId++) { + final long id = requestId; + final AtomicInteger cancelCount = new AtomicInteger(0); + final TestLocalRequest request = + new TestLocalRequest(id, newSocket(group)) { + public boolean tryCancel() { + cancelCount.incrementAndGet(); + return super.tryCancel(); + } + }; + final CyclicBarrier barrier = new CyclicBarrier(2); + Future registered = executor.submit(new Callable() { + public Boolean call() throws Exception { + barrier.await(); + return request.register(); + } + }); + Future cancelled = executor.submit(new Callable() { + public Object call() throws Exception { + barrier.await(); + return group.receiveRequestCancel(id); + } + }); + cancelled.get(); + Boolean live = registered.get(); + + Assert.assertTrue(request.isCancelled(), "round " + id + + ": cancel was lost"); + Assert.assertEquals(cancelCount.get(), 1, "round " + id + + ": tryCancel() delivered more than once"); + Assert.assertFalse(group.getLocalRequests().contains(id), "round " + id + + ": cancelled request left registered (register() == " + live + ")"); + } + } finally { + executor.shutdownNow(); + } + } +} diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java index 3b31cfc2..fba02bc0 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.common.rpc; @@ -73,7 +75,9 @@ public void onMessage(H socket, String message) { if (obj.request >= 0) { TestLocalRequest request = new TestLocalRequest(obj.messageId, socket); try { - getConnectionGroup().receiveRequest(request).execute(obj.request); + if (request.register()) { + request.execute(obj.request); + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); // request.handleException(e); diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java index 732d6072..6df8f967 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.common.rpc.impl; diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java index 7003b810..d3aeddfb 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java @@ -465,6 +465,7 @@ private static class InternalLocalOperationProcessor extends AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; private final AtomicBoolean commandChannelComplete = new AtomicBoolean(false); private final AtomicInteger resultChannelComplete = new AtomicInteger(0); @@ -553,6 +554,7 @@ public void onNext(BatchResult result) { } }; + Subscription batchSubscription = null; if (!requestMessage.getQuery()) { final List tasks = new ArrayList(); @@ -588,7 +590,7 @@ public void onNext(BatchResult result) { } try { - subscription = connectorFacade.executeBatch(tasks, observer, options); + batchSubscription = connectorFacade.executeBatch(tasks, observer, options); } catch (Throwable t) { tryHandleError(new ConnectorException(t)); return BatchOpResult.newBuilder().build(); @@ -599,16 +601,20 @@ public void onNext(BatchResult result) { for (int i = 0; i < requestMessage.getBatchTokenCount(); i++) { queryToken.addToken(requestMessage.getBatchToken(i)); } - subscription = connectorFacade.queryBatch(queryToken, observer, options); + batchSubscription = connectorFacade.queryBatch(queryToken, observer, options); } catch (Throwable t) { tryHandleError(new ConnectorException(t)); return BatchOpResult.newBuilder().build(); } } + if (null != batchSubscription) { + attachSubscription(batchSubscription); + } + BatchOpResult.Builder result = BatchOpResult.newBuilder(); - if (subscription != null && subscription.getReturnValue() != null) { - BatchToken returnedToken = (BatchToken) subscription.getReturnValue(); + if (batchSubscription != null && batchSubscription.getReturnValue() != null) { + BatchToken returnedToken = (BatchToken) batchSubscription.getReturnValue(); for (String token : returnedToken.getTokens()) { result.addBatchToken(token); } @@ -623,8 +629,35 @@ public void onNext(BatchResult result) { return result.build(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java index eb4c3006..0b731297 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework.async.impl; @@ -227,6 +229,7 @@ private static class InternalLocalOperationProcessor extends AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; protected InternalLocalOperationProcessor(long requestId, WebSocketConnectionHolder socket, ConnectorEventSubscriptionOpRequest message) { @@ -265,7 +268,7 @@ protected ConnectorEventSubscriptionOpResponse executeOperation( operationOptions = MessagesUtil.deserializeLegacy(requestMessage.getOptions()); } - subscription = + final Subscription result = connectorFacade.subscribe(objectClass, token, new Observer() { public void onCompleted() { @@ -294,11 +297,39 @@ public void onNext(ConnectorObject syncDelta) { }, operationOptions); + attachSubscription(result); return ConnectorEventSubscriptionOpResponse.getDefaultInstance(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java index 149cb207..8cd51ec6 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework.async.impl; @@ -219,6 +221,7 @@ private static class InternalLocalOperationProcessor AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; protected InternalLocalOperationProcessor(long requestId, WebSocketConnectionHolder socket, SyncEventSubscriptionOpRequest message) { @@ -258,7 +261,7 @@ protected SyncEventSubscriptionOpResponse executeOperation( operationOptions = MessagesUtil.deserializeLegacy(requestMessage.getOptions()); } - subscription = connectorFacade.subscribe(objectClass, token, new Observer() { + final Subscription result = connectorFacade.subscribe(objectClass, token, new Observer() { public void onCompleted() { handleResult(SyncEventSubscriptionOpResponse.newBuilder().setCompleted( @@ -283,11 +286,39 @@ public void onNext(SyncDelta syncDelta) { }, operationOptions); + attachSubscription(result); return SyncEventSubscriptionOpResponse.getDefaultInstance(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java index 9c3d9521..a868b67e 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java @@ -37,6 +37,7 @@ import org.forgerock.openicf.common.protobuf.RPCMessages.RemoteMessage; import org.forgerock.openicf.framework.ConnectorFramework; import org.forgerock.openicf.framework.async.AsyncConnectorInfoManager; +import org.forgerock.openicf.framework.async.impl.AbstractLocalOperationProcessor; import org.forgerock.openicf.framework.async.impl.AuthenticationAsyncApiOpImpl; import org.forgerock.openicf.framework.async.impl.BatchApiOpImpl; import org.forgerock.openicf.framework.async.impl.ConnectorEventSubscriptionApiOpImpl; @@ -423,56 +424,55 @@ public void processOperationRequest(final WebSocketConnectionHolder socket, ConnectorFacade connectorFacade = newInstance(socket, info, connectorFacadeKey); + AbstractLocalOperationProcessor processor = null; if (message.hasBatchOpRequest()) { - BatchApiOpImpl.createProcessor(messageId, socket, - message.getBatchOpRequest()).execute(connectorFacade); + processor = BatchApiOpImpl.createProcessor(messageId, socket, + message.getBatchOpRequest()); } else if (message.hasAuthenticateOpRequest()) { - AuthenticationAsyncApiOpImpl.createProcessor(messageId, socket, - message.getAuthenticateOpRequest()).execute(connectorFacade); + processor = AuthenticationAsyncApiOpImpl.createProcessor(messageId, socket, + message.getAuthenticateOpRequest()); } else if (message.hasCreateOpRequest()) { - CreateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getCreateOpRequest()).execute(connectorFacade); + processor = CreateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getCreateOpRequest()); } else if (message.hasConnectorEventSubscriptionOpRequest()) { - ConnectorEventSubscriptionApiOpImpl.createProcessor(messageId, socket, - message.getConnectorEventSubscriptionOpRequest()).execute( - connectorFacade); + processor = ConnectorEventSubscriptionApiOpImpl.createProcessor(messageId, + socket, message.getConnectorEventSubscriptionOpRequest()); } else if (message.hasDeleteOpRequest()) { - DeleteAsyncApiOpImpl.createProcessor(messageId, socket, - message.getDeleteOpRequest()).execute(connectorFacade); + processor = DeleteAsyncApiOpImpl.createProcessor(messageId, socket, + message.getDeleteOpRequest()); } else if (message.hasGetOpRequest()) { - GetAsyncApiOpImpl.createProcessor(messageId, socket, - message.getGetOpRequest()).execute(connectorFacade); + processor = GetAsyncApiOpImpl.createProcessor(messageId, socket, + message.getGetOpRequest()); } else if (message.hasResolveUsernameOpRequest()) { - ResolveUsernameAsyncApiOpImpl.createProcessor(messageId, socket, - message.getResolveUsernameOpRequest()).execute(connectorFacade); + processor = ResolveUsernameAsyncApiOpImpl.createProcessor(messageId, + socket, message.getResolveUsernameOpRequest()); } else if (message.hasSchemaOpRequest()) { - SchemaAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSchemaOpRequest()).execute(connectorFacade); + processor = SchemaAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSchemaOpRequest()); } else if (message.hasScriptOnConnectorOpRequest()) { - ScriptOnConnectorAsyncApiOpImpl.createProcessor(messageId, socket, - message.getScriptOnConnectorOpRequest()).execute(connectorFacade); + processor = ScriptOnConnectorAsyncApiOpImpl.createProcessor(messageId, + socket, message.getScriptOnConnectorOpRequest()); } else if (message.hasScriptOnResourceOpRequest()) { - ScriptOnResourceAsyncApiOpImpl.createProcessor(messageId, socket, - message.getScriptOnResourceOpRequest()).execute(connectorFacade); + processor = ScriptOnResourceAsyncApiOpImpl.createProcessor(messageId, + socket, message.getScriptOnResourceOpRequest()); } else if (message.hasSearchOpRequest()) { - SearchAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSearchOpRequest()).execute(connectorFacade); + processor = SearchAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSearchOpRequest()); } else if (message.hasSyncOpRequest()) { - SyncAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSyncOpRequest()).execute(connectorFacade); + processor = SyncAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSyncOpRequest()); } else if (message.hasSyncEventSubscriptionOpRequest()) { - SyncEventSubscriptionApiOpImpl.createProcessor(messageId, socket, - message.getSyncEventSubscriptionOpRequest()).execute( - connectorFacade); + processor = SyncEventSubscriptionApiOpImpl.createProcessor(messageId, + socket, message.getSyncEventSubscriptionOpRequest()); } else if (message.hasTestOpRequest()) { - TestAsyncApiOpImpl.createProcessor(messageId, socket, - message.getTestOpRequest()).execute(connectorFacade); + processor = TestAsyncApiOpImpl.createProcessor(messageId, socket, + message.getTestOpRequest()); } else if (message.hasUpdateOpRequest()) { - UpdateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getUpdateOpRequest()).execute(connectorFacade); + processor = UpdateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getUpdateOpRequest()); } else if (message.hasValidateOpRequest()) { - ValidateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getValidateOpRequest()).execute(connectorFacade); + processor = ValidateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getValidateOpRequest()); } else { socket.getRemoteConnectionContext().getRemoteConnectionGroup() .trySendMessage( @@ -480,6 +480,15 @@ public void processOperationRequest(final WebSocketConnectionHolder socket, new ConnectorException("Unknown OperationRequest")) .build()); } + + // Registration is separate from construction so a + // concurrently processed CancelOpRequest can never observe + // a partially constructed processor; register() returns + // false when a pending cancel already consumed the + // request, in which case the operation must not start. + if (null != processor && processor.register()) { + processor.execute(connectorFacade); + } } catch (Throwable t) { logger.ok(t, "Failed handle OperationRequest {0}", messageId); socket.getRemoteConnectionContext().getRemoteConnectionGroup().trySendMessage(