public ExecutionVenueNioServer createServer(TlsNioConfig cfg) {
    ExecutionVenueNioServer server = new ExecutionVenueNioServer();
    server.setNioConfig(cfg);
    NioLogger sessionLogger = new NioLogger("ALL");
    TransportCommandProcessor<SocketTransportCommand> processor = new SocketTransportCommandProcessor();
    CougarObjectIOFactory objectIOFactory = new HessianObjectIOFactory(false);
    ExecutionVenueServerHandler serverHandler = new ExecutionVenueServerHandler(sessionLogger, processor, objectIOFactory) {
        @Override
        public void messageReceived(IoSession session, Object message) throws Exception {
            session.write(message);
        }
    };
    server.setServerHandler(serverHandler);
    server.setServerExecutor(Executors.newCachedThreadPool());
    server.setSocketAcceptorProcessors(1);
    server.setTransportRegistry(new TransportRegistryImpl());
    final IoSessionManager sessionManager = new IoSessionManager();
    server.setSessionManager(sessionManager);
    sessionManager.setMaxTimeToWaitForRequestCompletion(5000);
    sessionManager.setNioLogger(sessionLogger);
    return server;
}
 

private void terminateAllSubscriptions(Subscription.CloseReason reason) {
    if (heapsByClient != null) {
        List<IoSession> sessions;
        try {
            subTermLock.lock();
            // take a copy in case it's being modified as we shutdown
            sessions = new ArrayList<IoSession>(heapsByClient.keySet());
        } finally {
            subTermLock.unlock();
        }

        for (IoSession session : sessions) {
            terminateSubscriptions(session, reason);
        }
    }
}
 

@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
    if (cause instanceof IOException) {
        ioExceptions.incrementAndGet();

        // We arrive here when the output pipe is broken. Broken network connections are not
        // really exceptional and should not be reported by dumping the stack trace.
        // Instead a summary debug level log message with some relevant info
        sessionLogger.log(ALL, session, "ExecutionVenueServerHandler: IOException received on session - closing");
    } else {
        otherExceptions.incrementAndGet();
        sessionLogger.log(SESSION, session, "ExecutionVenueServerHandler: Unexpected exception from session - see main log for details");
        LOG.warn("Unexpected exception from session " + NioUtils.getSessionId(session), cause);
    }
    session.close();
}
 

@Test
public void firstSubscriptionInitialUpdateReceivedLate() throws Exception {
    NewHeapSubscription newHeapSubscription = new NewHeapSubscription(1, "sub1", "firstSubscriptionInitialUpdateReceivedLate");
    InvocationResponse response = new InvocationResponseImpl(newHeapSubscription);
    WaitingObserver observer = new WaitingObserver();

    IoSession session = new MyIoSession(String.valueOf(ioSessionId++));
    subject.handleSubscriptionResponse(session, response, observer);

    assertTrue(observer.await(1000L));
    assertTrue(observer.getExecutionResult().isFault());

    HeapDelta delta = new HeapDelta(1, 0, createUpdateList(createInitial(new InstallRoot(0, NodeType.SCALAR), new SetScalar(0, true))));
    subject.applyDelta(session, delta);

    assertNull(subject.getHeapsByServer().get(NioUtils.getSessionId(session)));
}
 

public String addSubscription(LogExtension logExtension, Subscription subscription, IoSession session) {
    SubscriptionDetails details = new SubscriptionDetails();
    details.logExtension = logExtension;
    details.subscription = subscription;

    String id = uuidGenerator.getNextUUID();
    subscriptions.put(id, details);
    List<String> subscriptionIds = sessionSubscriptions.get(session);
    if (subscriptionIds == null) {
        subscriptionIds = new ArrayList<String>();
        sessionSubscriptions.put(session, subscriptionIds);
    }
    subscriptionIds.add(id);

    logSubscriptionStart(id, logExtension);

    return id;
}
 

@Test
public void responseReceivedAfterTimeout() throws IOException, InterruptedException {
    IoSession session = mock(IoSession.class);
    NioLogger logger = new NioLogger("ALL");
    RequestResponseManagerImpl impl = new RequestResponseManagerImpl(session, logger, 1);
    assertEquals(0, impl.getOutstandingRequestCount());

    WaitableResponseHandler responseHandler = new WaitableResponseHandler();
    long correlationId =  impl.sendRequest(new byte[0], responseHandler);

    Thread.sleep(2); // 2ms > 1ms

    impl.checkForExpiredRequests();

    ResponseMessage message = new ResponseMessage(correlationId, new byte[0]);
    impl.messageReceived(session, message);
    // just want no exceptions
}
 

protected boolean doDecode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception {
    if (in.remaining() >= PAYLOAD_SIZE) {
        byte[] buf = new byte[in.remaining()];
        in.get(buf);
        
        // first 7 bytes are the sensor ID, last is the status
        // and the result message will look something like
        // MachineID=2371748;Status=Good
        StringBuilder sb = new StringBuilder();
        sb.append("MachineID=")
        .append(new String(buf, 0, PAYLOAD_SIZE - 1)).append(";")
        .append("Status=");
        if (buf[PAYLOAD_SIZE - 1] == '1') {
            sb.append("Good");
        } else {
            sb.append("Failure");
        }
        out.write(sb.toString());
        return true;
    } else {
        return false;
    }
}
 

@Override
public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
    // if we want to terminate based on queue depth then check it..
    SessionWriteQueueMonitor monitor = monitors.get(NioUtils.getSessionId(session));
    if (monitor != null) {
        long newDepth = monitor.countIn();
        if (maxWriteQueueSize > 0 && newDepth > maxWriteQueueSize) {
            logger.log(NioLogger.LoggingLevel.SESSION, session, "Session exceeded max writeQueue size of %s, closing session", maxWriteQueueSize);
            // kill
            session.close();
            return;
        }
    }

    nextFilter.filterWrite(session, writeRequest);
}
 

@Test
public void testRejectionServerUnhealthy() throws IOException {
    ExecutionVenueNioServer server = createServer(defaultServerConfig);
    server.start();
    server.setHealthState(false);

    IoSession session = createClient(defaultClientConfig, new IoHandlerAdapter());

    ClientHandshake handshake = (ClientHandshake) session.getAttribute(ClientHandshake.HANDSHAKE);
    handshake.await(10000);

    boolean success = handshake.successful();

    session.close();
    server.stop();

    assertEquals("connection shouln't have been successful", false, success);
}
 

@Override
public void exceptionCaught(IoSession session, Throwable cause) {
    if (cause instanceof IOException) {
        LOG.debug("IO exception from session "+NioUtils.getSessionId(session), cause);
    } else {
        LOG.warn("Unexpected exception from session "+NioUtils.getSessionId(session), cause);
    }
    nioLogger.log(NioLogger.LoggingLevel.SESSION, session, "RequestResponseManager - %s received: %s - closing session", cause.getClass().getSimpleName(), cause.getMessage());
    session.close();
}
 

private MinaChannel(IoSession session, URL url, ChannelHandler handler) {
    super(url, handler);
    if (session == null) {
        throw new IllegalArgumentException("mina session == null");
    }
    this.session = session;
}
 

public static void writeEventMessageToSession(IoSession session, Object obj, CougarObjectIOFactory objectIOFactory) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    CougarObjectOutput out = objectIOFactory.newCougarObjectOutput(baos, CougarProtocol.getProtocolVersion(session));
    out.writeObject(obj);
    out.flush();

    session.write(new EventMessage(baos.toByteArray()));
}
 

public static IoSession newSession(byte version) {
    final IoSession mockSession = mock(IoSession.class);
    when(mockSession.getAttribute(CougarProtocol.PROTOCOL_VERSION_ATTR_NAME)).thenReturn(version);
    final WriteFuture mockFuture = mock(WriteFuture.class);
    when(mockSession.write(any())).thenReturn(mockFuture);
    return mockSession;
}
 

@Override
public void messageSent(IoSession session, Object message) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.sent(channel, message);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
}
 

@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.received(channel, message);
    } finally {
        MinaChannel.removeChannelIfDisconnected(session);
    }
}
 

@Override
public void messageSent(IoSession session, Object message) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.sent(channel, message);
    } finally {
        MinaChannel.removeChannelIfDisconnected(session);
    }
}
 

public List<String> getHeapsForSession(IoSession session) {
    List<String> ret = new ArrayList<String>();
    try {
        subTermLock.lock();
        Multiset<String> s = heapsByClient.get(session);
        if (s != null) {
            ret.addAll(s.keySet());
        }
    } finally {
        subTermLock.unlock();
    }
    return ret;
}
 

@Override
protected Channel getChannel() {
    IoSession s = session;
    if (s == null || !s.isConnected())
        return null;
    return MinaChannel.getOrAddChannel(s, getUrl(), this);
}
 

public Channel getChannel(InetSocketAddress remoteAddress) {
    Set<IoSession> sessions = acceptor.getManagedSessions(getBindAddress());
    for (IoSession session : sessions) {
        if (session.getRemoteAddress().equals(remoteAddress)) {
            return MinaChannel.getOrAddChannel(session, getUrl(), this);
        }
    }
    return null;
}
 

@Override
public void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception {
    if (!isServer) {
        ClientHandshake clientHandshake = new ClientHandshake();
        session.setAttribute(ClientHandshake.HANDSHAKE, clientHandshake);
        session.write(new ConnectMessage(getClientAcceptableVersions()));
    }
    super.sessionOpened(nextFilter, session);
}
 

public void encode(IoSession ioSession, Object message, ProtocolEncoderOutput out)
    throws Exception {             
    ByteBuffer buf = ByteBuffer.allocate(PAYLOAD_SIZE);
    String s = (String) message;                    
    buf.put(s.getBytes());
    buf.flip();
    out.write(buf);
}
 

public void encode(IoSession session, Object msg, ProtocolEncoderOutput out) throws Exception {
    ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(1024);
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
    	codec.encode(channel, buffer, msg);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
    out.write(ByteBuffer.wrap(buffer.toByteBuffer()));
    out.flush();
}
 

@Override
public void sessionOpened(IoSession session) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.connected(channel);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
}
 

@Test
public void testSessionsAreClosedOnlyAfterOutstandingRequestsAreServed() {
    final IoSession ioSession = newSession(V2);
    ExecutionVenueServerHandler serverHandler = mock(ExecutionVenueServerHandler.class);
    when(serverHandler.getOutstandingRequests()).thenReturn(2l, 1l, 0l); // Counting down

    sessionManager.shutdownSessions(singleton(ioSession), cougarProtocol, serverHandler);

    final InOrder inOrder = inOrder(ioSession, serverHandler);
    inOrder.verify(ioSession).write(isA(SuspendMessage.class));
    inOrder.verify(serverHandler, times(3)).getOutstandingRequests();
    inOrder.verify(ioSession).write(isA(DisconnectMessage.class));
}
 

@Override
public void messageSent(IoSession session, Object message) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.sent(channel, message);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
}
 

@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.caught(channel, cause);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
}
 

@Test
public void testSuspendMessagesAreWrittenForV2Sessions() {
    CougarProtocol protocol = CougarProtocol.getServerInstance(new NioLogger("NONE"), 5000, 5000, null, false, false);
    IoSession ioSession = newV2Session();
    protocol.suspendSession(ioSession);
    verify(ioSession).write(isA(SuspendMessage.class));
}
 

@Before
public void setup() throws Exception {
    sessionFactory = mock(IoSessionFactory.class);

    Field sessionsField = IoSessionFactory.class.getDeclaredField("sessions");
    sessionsField.setAccessible(true);
    Map<SocketAddress, IoSession> sessions = new HashMap<SocketAddress, IoSession>();
    final IoSession session1 = getSession(1);
    final IoSession session2 = getSession(2);
    final IoSession session3 = getSession(3);
    sessions.put(session1.getRemoteAddress(), session1);
    sessions.put(session2.getRemoteAddress(), session2);
    sessions.put(session3.getRemoteAddress(), session3);

    sessionsField.set(sessionFactory, sessions);
    when(sessionFactory.getSession()).thenCallRealMethod();
    when(sessionFactory.isAvailable(any(IoSession.class))).thenCallRealMethod();

    Field lockField = IoSessionFactory.class.getDeclaredField("lock");
    lockField.setAccessible(true);
    lockField.set(sessionFactory, new Object());

    NetworkAddressResolver resolver = mock(NetworkAddressResolver.class);
    when(resolver.resolve(HOST1)).thenReturn(asSet(HOST1_IP1, HOST1_IP2));
    when(resolver.resolve(HOST2)).thenReturn(asSet(HOST2_IP1, HOST2_IP2));
    when(resolver.resolve(HOST3)).thenReturn(asSet(HOST3_IP1, HOST3_IP2));
    hosts = "HOST1:9003,HOST2:9003,HOST3:9003";
    recycler = new SessionRecycler(sessionFactory, resolver, hosts, 5000);
}
 

@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
    try {
        handler.received(channel, message);
    } finally {
        MinaChannel.removeChannelIfDisconnectd(session);
    }
}
 

@Override
protected Channel getChannel() {
    IoSession s = session;
    if (s == null || ! s.isConnected())
        return null;
    return MinaChannel.getOrAddChannel(s, getUrl(), this);
}