@Test public void addSecondSubscriptionMidStream() throws Exception { SocketTransportCommandProcessor commandProcessor = mock(SocketTransportCommandProcessor.class); when(commandProcessor.writeSuccessResponse(any(SocketTransportRPCCommand.class), any(ExecutionResult.class), any(DehydratedExecutionContext.class))).thenReturn(true); SocketTransportRPCCommand command = mock(SocketTransportRPCCommand.class); DehydratedExecutionContext requestContext = mock(DehydratedExecutionContext.class); MutableHeap heap = new MutableHeap("addSecondSubscriptionMidStream"); Subscription sub = mock(Subscription.class); ConnectedResponse subscriptionResult = new ConnectedResponseImpl(heap, sub); OperationDefinition operationDefinition = mock(OperationDefinition.class); List<Update> expectedUpdates = new ArrayList<Update>(); expectedUpdates.add(createInitial(new InstallRoot(0, NodeType.OBJECT), new InstallField(0, 1, "value", NodeType.SCALAR), new SetScalar(1, 1))); expectedUpdates.add(createUpdate(new SetScalar(1, 2))); expectedUpdates.add(createInitial(new InstallRoot(0, NodeType.OBJECT), new InstallField(0, 1, "value", NodeType.SCALAR), new SetScalar(1, 2))); expectedUpdates.add(createUpdate(new SetScalar(1, 3))); cougarOutput.setExpectedUpdates(expectedUpdates); heap.beginUpdate(); SimpleConnectedObject object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(1); heap.endUpdate(); MyIoSession session = new MyIoSession(String.valueOf(ioSessionId++)); session.setAttribute(CougarProtocol.PROTOCOL_VERSION_ATTR_NAME, CougarProtocol.TRANSPORT_PROTOCOL_VERSION_MAX_SUPPORTED); when(command.getSession()).thenReturn(session); subject.addSubscription(commandProcessor, command, subscriptionResult, operationDefinition, requestContext, null); heap.beginUpdate(); object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(2); heap.endUpdate(); MyIoSession session2 = new MyIoSession(String.valueOf(ioSessionId++)); session2.setAttribute(CougarProtocol.PROTOCOL_VERSION_ATTR_NAME, CougarProtocol.TRANSPORT_PROTOCOL_VERSION_MAX_SUPPORTED); when(command.getSession()).thenReturn(session2); subject.addSubscription(commandProcessor, command, subscriptionResult, operationDefinition, requestContext, null); heap.beginUpdate(); object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(3); heap.endUpdate(); assertExpectedUpdatesWritten(); int updatesWritten = cougarOutput.getAllValues().size() - 2; assertExpectedSessionWrites(session, updatesWritten + 1); assertExpectedSessionWrites(session2, updatesWritten); verify(sub, never()).close(); verify(sub, never()).close(any(Subscription.CloseReason.class)); }
@Override public void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension) { final String heapUri = result.getHeap().getUri(); HeapState heapState = null; try { boolean readyToContinue = false; while (!readyToContinue) { subTermLock.lock(); heapState = heapStates.get(heapUri); boolean wasNewHeapState = heapState == null; if (wasNewHeapState) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { subTermLock.unlock(); heapState.getUpdateLock().lock(); subTermLock.lock(); if (!heapStates.containsKey(heapUri)) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { HeapState latestState = heapStates.get(heapUri); if (latestState.getInstanceId() == heapState.getInstanceId()) { readyToContinue = true; } else { heapState.getUpdateLock().unlock(); heapState = latestState; } } } } final HeapState finalHeapState = heapState; final Subscription subscription = result.getSubscription(); result.getHeap().traverse(new UpdateProducingHeapListener() { @Override protected void doUpdate(Update u) { boolean updateContainsTermination = u.getActions().contains(TerminateHeap.INSTANCE); if (updateContainsTermination) { terminateSubscriptions(command.getSession(), heapUri, REQUESTED_BY_PUBLISHER); commandProcessor.writeErrorResponse(command, context, new CougarFrameworkException("Subscription requested for terminated heap: " + heapUri), true); return; } Multiset<String> heapsForThisClient = heapsByClient.get(command.getSession()); if (heapsForThisClient == null) { heapsForThisClient = new Multiset<String>(); heapsByClient.put(command.getSession(), heapsForThisClient); } long heapId = finalHeapState.getHeapId(); final String subscriptionId = finalHeapState.addSubscription(connectedObjectLogExtension, subscription, command.getSession()); subscription.addListener(new Subscription.SubscriptionListener() { @Override public void subscriptionClosed(Subscription subscription, Subscription.CloseReason reason) { if (reason == REQUESTED_BY_PUBLISHER) { PooledServerConnectedObjectManager.this.terminateSubscription(command.getSession(), heapUri, subscriptionId, reason); } finalHeapState.logSubscriptionEnd(subscriptionId, connectedObjectLogExtension, reason); } }); boolean newHeapDefinition = heapsForThisClient.count(heapUri) == 0; heapsForThisClient.add(heapUri); NewHeapSubscription response; if (newHeapDefinition) { response = new NewHeapSubscription(heapId, subscriptionId, heapUri); } else { response = new NewHeapSubscription(heapId, subscriptionId); } ExecutionResult executionResult = new ExecutionResult(response); boolean successfulResponse = commandProcessor.writeSuccessResponse(command, executionResult, context); if (!successfulResponse) { terminateSubscriptions(command.getSession(), heapUri, INTERNAL_ERROR); } if (newHeapDefinition) { finalHeapState.getQueuedChanges().add(new QueuedHeapChange(new QueuedSubscription(command.getSession(), new InitialUpdate(u)))); heapsWaitingForUpdate.add(heapUri); } } }); } finally { subTermLock.unlock(); assert heapState != null; heapState.getUpdateLock().unlock(); } }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension) { final String heapUri = result.getHeap().getUri(); HeapState heapState = null; try { boolean readyToContinue = false; while (!readyToContinue) { subTermLock.lock(); heapState = heapStates.get(heapUri); boolean wasNewHeapState = heapState == null; if (wasNewHeapState) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { subTermLock.unlock(); heapState.getUpdateLock().lock(); subTermLock.lock(); if (!heapStates.containsKey(heapUri)) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { HeapState latestState = heapStates.get(heapUri); if (latestState.getInstanceId() == heapState.getInstanceId()) { readyToContinue = true; } else { heapState.getUpdateLock().unlock(); heapState = latestState; } } } } final HeapState finalHeapState = heapState; final Subscription subscription = result.getSubscription(); result.getHeap().traverse(new UpdateProducingHeapListener() { @Override protected void doUpdate(Update u) { boolean updateContainsTermination = u.getActions().contains(TerminateHeap.INSTANCE); if (updateContainsTermination) { terminateSubscriptions(command.getSession(), heapUri, REQUESTED_BY_PUBLISHER); commandProcessor.writeErrorResponse(command, context, new CougarFrameworkException("Subscription requested for terminated heap: " + heapUri), true); return; } Multiset<String> heapsForThisClient = heapsByClient.get(command.getSession()); if (heapsForThisClient == null) { heapsForThisClient = new Multiset<String>(); heapsByClient.put(command.getSession(), heapsForThisClient); } long heapId = finalHeapState.getHeapId(); final String subscriptionId = finalHeapState.addSubscription(connectedObjectLogExtension, subscription, command.getSession()); subscription.addListener(new Subscription.SubscriptionListener() { @Override public void subscriptionClosed(Subscription subscription, Subscription.CloseReason reason) { if (reason == REQUESTED_BY_PUBLISHER) { PooledServerConnectedObjectManager.this.terminateSubscription(command.getSession(), heapUri, subscriptionId, reason); } finalHeapState.logSubscriptionEnd(subscriptionId, connectedObjectLogExtension, reason); } }); boolean newHeapDefinition = heapsForThisClient.count(heapUri) == 0; heapsForThisClient.add(heapUri); NewHeapSubscription response; if (newHeapDefinition) { response = new NewHeapSubscription(heapId, subscriptionId, heapUri); } else { response = new NewHeapSubscription(heapId, subscriptionId); } ExecutionResult executionResult = new ExecutionResult(response); boolean successfulResponse = commandProcessor.writeSuccessResponse(command, executionResult, context); if (!successfulResponse) { terminateSubscriptions(command.getSession(), heapUri, INTERNAL_ERROR); } if (newHeapDefinition) { finalHeapState.getQueuedChanges().add(new QueuedHeapChange(new QueuedSubscription(command.getSession(), new InitialUpdate(u)))); heapsWaitingForUpdate.add(heapUri); } } }); } finally { subTermLock.unlock(); assert heapState != null; heapState.getUpdateLock().unlock(); } } }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension) { final String heapUri = result.getHeap().getUri(); HeapState heapState = null; try { boolean readyToContinue = false; while (!readyToContinue) { subTermLock.lock(); heapState = heapStates.get(heapUri); boolean wasNewHeapState = heapState == null; if (wasNewHeapState) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { subTermLock.unlock(); heapState.getUpdateLock().lock(); subTermLock.lock(); if (!heapStates.containsKey(heapUri)) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { HeapState latestState = heapStates.get(heapUri); if (latestState.getInstanceId() == heapState.getInstanceId()) { readyToContinue = true; } else { heapState.getUpdateLock().unlock(); heapState = latestState; } } } } final HeapState finalHeapState = heapState; final Subscription subscription = result.getSubscription(); result.getHeap().traverse(new UpdateProducingHeapListener() { @Override protected void doUpdate(Update u) { boolean updateContainsTermination = u.getActions().contains(TerminateHeap.INSTANCE); if (updateContainsTermination) { terminateSubscriptions(command.getSession(), heapUri, REQUESTED_BY_PUBLISHER); commandProcessor.writeErrorResponse(command, context, new CougarFrameworkException("Subscription requested for terminated heap: " + heapUri), true); return; } Multiset<String> heapsForThisClient = heapsByClient.get(command.getSession()); if (heapsForThisClient == null) { heapsForThisClient = new Multiset<String>(); heapsByClient.put(command.getSession(), heapsForThisClient); } long heapId = finalHeapState.getHeapId(); final String subscriptionId = finalHeapState.addSubscription(connectedObjectLogExtension, subscription, command.getSession()); subscription.addListener(new Subscription.SubscriptionListener() { @Override public void subscriptionClosed(Subscription subscription, Subscription.CloseReason reason) { if (reason == REQUESTED_BY_PUBLISHER) { PooledServerConnectedObjectManager.this.terminateSubscription(command.getSession(), heapUri, subscriptionId, reason); } finalHeapState.logSubscriptionEnd(subscriptionId, connectedObjectLogExtension, reason); } }); boolean newHeapDefinition = heapsForThisClient.count(heapUri) == 0; heapsForThisClient.add(heapUri); NewHeapSubscription response; if (newHeapDefinition) { response = new NewHeapSubscription(heapId, subscriptionId, heapUri); } else { response = new NewHeapSubscription(heapId, subscriptionId); } ExecutionResult executionResult = new ExecutionResult(response); boolean successfulResponse = commandProcessor.writeSuccessResponse(command, executionResult, context); if (!successfulResponse) { terminateSubscriptions(command.getSession(), heapUri, INTERNAL_ERROR); } if (newHeapDefinition) { finalHeapState.getQueuedChanges().add(new QueuedHeapChange(new QueuedSubscription(command.getSession(), new InitialUpdate(u)))); heapsWaitingForUpdate.add(heapUri); } } }); } finally { subTermLock.unlock(); assert heapState != null; heapState.getUpdateLock().unlock(); } }  }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension) { final String heapUri = result.getHeap().getUri(); HeapState heapState = null; try { boolean readyToContinue = false; while (!readyToContinue) { subTermLock.lock(); heapState = heapStates.get(heapUri); boolean wasNewHeapState = heapState == null; if (wasNewHeapState) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { subTermLock.unlock(); heapState.getUpdateLock().lock(); subTermLock.lock(); if (!heapStates.containsKey(heapUri)) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { HeapState latestState = heapStates.get(heapUri); if (latestState.getInstanceId() == heapState.getInstanceId()) { readyToContinue = true; } else { heapState.getUpdateLock().unlock(); heapState = latestState; } } } } final HeapState finalHeapState = heapState; final Subscription subscription = result.getSubscription(); result.getHeap().traverse(new UpdateProducingHeapListener() { @Override protected void doUpdate(Update u) { boolean updateContainsTermination = u.getActions().contains(TerminateHeap.INSTANCE); if (updateContainsTermination) { terminateSubscriptions(command.getSession(), heapUri, REQUESTED_BY_PUBLISHER); commandProcessor.writeErrorResponse(command, context, new CougarFrameworkException("Subscription requested for terminated heap: " + heapUri), true); return; } Multiset<String> heapsForThisClient = heapsByClient.get(command.getSession()); if (heapsForThisClient == null) { heapsForThisClient = new Multiset<String>(); heapsByClient.put(command.getSession(), heapsForThisClient); } long heapId = finalHeapState.getHeapId(); final String subscriptionId = finalHeapState.addSubscription(connectedObjectLogExtension, subscription, command.getSession()); subscription.addListener(new Subscription.SubscriptionListener() { @Override public void subscriptionClosed(Subscription subscription, Subscription.CloseReason reason) { if (reason == REQUESTED_BY_PUBLISHER) { PooledServerConnectedObjectManager.this.terminateSubscription(command.getSession(), heapUri, subscriptionId, reason); } finalHeapState.logSubscriptionEnd(subscriptionId, connectedObjectLogExtension, reason); } }); boolean newHeapDefinition = heapsForThisClient.count(heapUri) == 0; heapsForThisClient.add(heapUri); NewHeapSubscription response; if (newHeapDefinition) { response = new NewHeapSubscription(heapId, subscriptionId, heapUri); } else { response = new NewHeapSubscription(heapId, subscriptionId); } ExecutionResult executionResult = new ExecutionResult(response); boolean successfulResponse = commandProcessor.writeSuccessResponse(command, executionResult, context); if (!successfulResponse) { terminateSubscriptions(command.getSession(), heapUri, INTERNAL_ERROR); } if (newHeapDefinition) { finalHeapState.getQueuedChanges().add(new QueuedHeapChange(new QueuedSubscription(command.getSession(), new InitialUpdate(u)))); heapsWaitingForUpdate.add(heapUri); } } }); } finally { subTermLock.unlock(); assert heapState != null; heapState.getUpdateLock().unlock(); } }  Map<Long, String> getHeapUris(); List<String> getHeapsForSession(IoSession session); HeapStateMonitoring getHeapStateForMonitoring(String uri); @Override void setNioLogger(NioLogger nioLogger); void setObjectIOFactory(CougarObjectIOFactory objectIOFactory); void setNumProcessingThreads(int i); void setEventLogger(EventLogger eventLogger); void setMaxUpdateActionsPerMessage(int maxUpdateActionsPerMessage); void start(); void stop(); @Override void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension); @Override void terminateSubscription(IoSession session, TerminateSubscription payload); void terminateSubscription(IoSession session, String heapUri, String subscriptionId, Subscription.CloseReason reason); @Override void sessionOpened(IoSession session); @Override void sessionClosed(IoSession session); @ManagedAttribute(description = "Number of active heaps") int getNumberOfHeaps(); @ManagedOperation(description = "Number of subscriptions for the given heap URI") int getHeapSubscriptionCount(String heapUri); @ManagedOperation(description = "Number of sessions subscribed for the given heap URI") int getHeapSessionCount(String heapUri); @ManagedOperation(description = "Has the specified heap terminated") boolean hasHeapTerminated(String heapUri); @ManagedOperation(description = "Last received update Id") long getLastUpdateId(String heapUri); @ManagedOperation(description = "Number of updates queued for the specified heap") long showNumOfQueuedChanges(String heapUri); @ManagedAttribute(description = "Number of pusher threads") int getNumProcessingThreads(); }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension) { final String heapUri = result.getHeap().getUri(); HeapState heapState = null; try { boolean readyToContinue = false; while (!readyToContinue) { subTermLock.lock(); heapState = heapStates.get(heapUri); boolean wasNewHeapState = heapState == null; if (wasNewHeapState) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { subTermLock.unlock(); heapState.getUpdateLock().lock(); subTermLock.lock(); if (!heapStates.containsKey(heapUri)) { heapState = processHeapStateCreation(result, heapUri); readyToContinue = true; } else { HeapState latestState = heapStates.get(heapUri); if (latestState.getInstanceId() == heapState.getInstanceId()) { readyToContinue = true; } else { heapState.getUpdateLock().unlock(); heapState = latestState; } } } } final HeapState finalHeapState = heapState; final Subscription subscription = result.getSubscription(); result.getHeap().traverse(new UpdateProducingHeapListener() { @Override protected void doUpdate(Update u) { boolean updateContainsTermination = u.getActions().contains(TerminateHeap.INSTANCE); if (updateContainsTermination) { terminateSubscriptions(command.getSession(), heapUri, REQUESTED_BY_PUBLISHER); commandProcessor.writeErrorResponse(command, context, new CougarFrameworkException("Subscription requested for terminated heap: " + heapUri), true); return; } Multiset<String> heapsForThisClient = heapsByClient.get(command.getSession()); if (heapsForThisClient == null) { heapsForThisClient = new Multiset<String>(); heapsByClient.put(command.getSession(), heapsForThisClient); } long heapId = finalHeapState.getHeapId(); final String subscriptionId = finalHeapState.addSubscription(connectedObjectLogExtension, subscription, command.getSession()); subscription.addListener(new Subscription.SubscriptionListener() { @Override public void subscriptionClosed(Subscription subscription, Subscription.CloseReason reason) { if (reason == REQUESTED_BY_PUBLISHER) { PooledServerConnectedObjectManager.this.terminateSubscription(command.getSession(), heapUri, subscriptionId, reason); } finalHeapState.logSubscriptionEnd(subscriptionId, connectedObjectLogExtension, reason); } }); boolean newHeapDefinition = heapsForThisClient.count(heapUri) == 0; heapsForThisClient.add(heapUri); NewHeapSubscription response; if (newHeapDefinition) { response = new NewHeapSubscription(heapId, subscriptionId, heapUri); } else { response = new NewHeapSubscription(heapId, subscriptionId); } ExecutionResult executionResult = new ExecutionResult(response); boolean successfulResponse = commandProcessor.writeSuccessResponse(command, executionResult, context); if (!successfulResponse) { terminateSubscriptions(command.getSession(), heapUri, INTERNAL_ERROR); } if (newHeapDefinition) { finalHeapState.getQueuedChanges().add(new QueuedHeapChange(new QueuedSubscription(command.getSession(), new InitialUpdate(u)))); heapsWaitingForUpdate.add(heapUri); } } }); } finally { subTermLock.unlock(); assert heapState != null; heapState.getUpdateLock().unlock(); } }  Map<Long, String> getHeapUris(); List<String> getHeapsForSession(IoSession session); HeapStateMonitoring getHeapStateForMonitoring(String uri); @Override void setNioLogger(NioLogger nioLogger); void setObjectIOFactory(CougarObjectIOFactory objectIOFactory); void setNumProcessingThreads(int i); void setEventLogger(EventLogger eventLogger); void setMaxUpdateActionsPerMessage(int maxUpdateActionsPerMessage); void start(); void stop(); @Override void addSubscription(final SocketTransportCommandProcessor commandProcessor, final SocketTransportRPCCommand command, final ConnectedResponse result, final OperationDefinition operationDefinition, final DehydratedExecutionContext context, final LogExtension connectedObjectLogExtension); @Override void terminateSubscription(IoSession session, TerminateSubscription payload); void terminateSubscription(IoSession session, String heapUri, String subscriptionId, Subscription.CloseReason reason); @Override void sessionOpened(IoSession session); @Override void sessionClosed(IoSession session); @ManagedAttribute(description = "Number of active heaps") int getNumberOfHeaps(); @ManagedOperation(description = "Number of subscriptions for the given heap URI") int getHeapSubscriptionCount(String heapUri); @ManagedOperation(description = "Number of sessions subscribed for the given heap URI") int getHeapSessionCount(String heapUri); @ManagedOperation(description = "Has the specified heap terminated") boolean hasHeapTerminated(String heapUri); @ManagedOperation(description = "Last received update Id") long getLastUpdateId(String heapUri); @ManagedOperation(description = "Number of updates queued for the specified heap") long showNumOfQueuedChanges(String heapUri); @ManagedAttribute(description = "Number of pusher threads") int getNumProcessingThreads();  }