@Test public void sessionClosed() 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); MyIoSession session = new MyIoSession(String.valueOf(ioSessionId++)) { @Override public CloseFuture close() { subject.sessionClosed(this); return super.close(); } }; session.setAttribute(CougarProtocol.PROTOCOL_VERSION_ATTR_NAME, CougarProtocol.TRANSPORT_PROTOCOL_VERSION_MAX_SUPPORTED); when(command.getSession()).thenReturn(session); DehydratedExecutionContext requestContext = mock(DehydratedExecutionContext.class); MutableHeap heap = new MutableHeap("sessionClosed"); Subscription subscription = mock(Subscription.class); ConnectedResponse subscriptionResult = new ConnectedResponseImpl(heap, subscription); OperationDefinition operationDefinition = mock(OperationDefinition.class); subject.addSubscription(commandProcessor, command, subscriptionResult, operationDefinition, requestContext, null); heap.beginUpdate(); SimpleConnectedObject object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(1); heap.endUpdate(); heap.beginUpdate(); object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(2); heap.endUpdate(); session.close(); heap.beginUpdate(); object = objectProjector(SimpleConnectedObject.class).project(heap.ensureRoot(NodeType.OBJECT)); object.value().set(3); heap.endUpdate(); assertNull(subject.getHeapsByClient().get(session)); assertNull(subject.getHeapStates().get("sessionClosed")); verify(subscription).close(Subscription.CloseReason.CONNECTION_CLOSED); assertEquals(0, getHeapListeners(heap).size()); }
@Override public void sessionClosed(IoSession session) { nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Session closed, terminating live subscriptions"); terminateSubscriptions(session, CONNECTION_CLOSED); }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void sessionClosed(IoSession session) { nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Session closed, terminating live subscriptions"); terminateSubscriptions(session, CONNECTION_CLOSED); } }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void sessionClosed(IoSession session) { nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Session closed, terminating live subscriptions"); terminateSubscriptions(session, CONNECTION_CLOSED); }  }
PooledServerConnectedObjectManager implements ServerConnectedObjectManager { @Override public void sessionClosed(IoSession session) { nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Session closed, terminating live subscriptions"); terminateSubscriptions(session, CONNECTION_CLOSED); }  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 sessionClosed(IoSession session) { nioLogger.log(NioLogger.LoggingLevel.TRANSPORT, session, "Session closed, terminating live subscriptions"); terminateSubscriptions(session, CONNECTION_CLOSED); }  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();  }