@Test public void filterWriteShouldSwallowMessageIfCloseSent() throws Exception { Mockery context = new Mockery() {{ setImposteriser(ClassImposteriser.INSTANCE); }}; context.setThreadingPolicy(new Synchroniser()); final NextFilter nextFilter = context.mock(NextFilter.class); final IoSession session = context.mock(IoSession.class); final WriteRequest closeRequest = context.mock(WriteRequest.class, "closeRequest"); final WriteRequest writeRequest = context.mock(WriteRequest.class); final Logger logger = context.mock(Logger.class); final ScheduledExecutorService scheduler = context.mock(ScheduledExecutorService.class); final WriteFuture closeFuture = context.mock(WriteFuture.class, "closeFuture"); final WsMessage close = new WsCloseMessage(); context.checking(new Expectations() { { allowing(logger).isTraceEnabled(); will(returnValue(false)); oneOf(session).isConnected(); will(returnValue(true)); oneOf(closeRequest).getFuture(); will(returnValue(closeFuture)); oneOf(closeFuture).addListener(with(any(IoFutureListener.class))); exactly(2).of(closeRequest).getMessage(); will(returnValue(close)); oneOf(nextFilter).filterWrite(session, closeRequest); oneOf(scheduler).schedule(with(any(WsCloseFilter.class)), with(any(Long.class)), with(TimeUnit.MILLISECONDS)); } }); Properties configuration = new Properties(); WsCloseFilter filter = new WsCloseFilter(WebSocketWireProtocol.RFC_6455, configuration, logger, scheduler); filter.filterWrite(nextFilter, session, closeRequest); filter.filterWrite(nextFilter, session, writeRequest); context.assertIsSatisfied(); }
@Override public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { if (sentCloseFrame.get()) { return; } super.filterWrite(nextFilter, session, writeRequest); }
WsCloseFilter extends WsFilterAdapter implements Runnable { @Override public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { if (sentCloseFrame.get()) { return; } super.filterWrite(nextFilter, session, writeRequest); } }
WsCloseFilter extends WsFilterAdapter implements Runnable { @Override public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { if (sentCloseFrame.get()) { return; } super.filterWrite(nextFilter, session, writeRequest); } WsCloseFilter(WebSocketWireProtocol wsVersion,
                         Properties configuration,
                         Logger logger,
                         ScheduledExecutorService scheduler); }
WsCloseFilter extends WsFilterAdapter implements Runnable { @Override public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { if (sentCloseFrame.get()) { return; } super.filterWrite(nextFilter, session, writeRequest); } WsCloseFilter(WebSocketWireProtocol wsVersion,
                         Properties configuration,
                         Logger logger,
                         ScheduledExecutorService scheduler); @Override void filterWrite(NextFilter nextFilter, IoSession session,
            WriteRequest writeRequest); @Override void messageReceived(NextFilter nextFilter,
                                IoSession session,
                                Object message); @Override void filterClose(final NextFilter nextFilter,
                            final IoSession session); @Override void onPreRemove(IoFilterChain filterChain,
                            String name,
                            NextFilter nextFilter); @Override void run(); static boolean neededForProtocolVersion(WebSocketWireProtocol wsVersion); }
WsCloseFilter extends WsFilterAdapter implements Runnable { @Override public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception { if (sentCloseFrame.get()) { return; } super.filterWrite(nextFilter, session, writeRequest); } WsCloseFilter(WebSocketWireProtocol wsVersion,
                         Properties configuration,
                         Logger logger,
                         ScheduledExecutorService scheduler); @Override void filterWrite(NextFilter nextFilter, IoSession session,
            WriteRequest writeRequest); @Override void messageReceived(NextFilter nextFilter,
                                IoSession session,
                                Object message); @Override void filterClose(final NextFilter nextFilter,
                            final IoSession session); @Override void onPreRemove(IoFilterChain filterChain,
                            String name,
                            NextFilter nextFilter); @Override void run(); static boolean neededForProtocolVersion(WebSocketWireProtocol wsVersion);  }