@Test public void shouldCloseWsnSessionWhenTransportClosesCleanlyButUnexpectedly() throws Exception { String location = "wsn: Map<String, Object> addressOptions = Collections.emptyMap(); ResourceAddress address = addressFactory.newResourceAddress(location, addressOptions); final CountDownLatch waitForClientParentSessionCloseListenerEstabished = new CountDownLatch(1); IoHandler acceptHandler = new IoHandlerAdapter<IoSessionEx>() { @Override protected void doMessageReceived(IoSessionEx session, Object message) throws Exception { IoBuffer buf = (IoBuffer)message; WriteFuture future = session.write(buf.duplicate()); future.addListener(new IoFutureListener<WriteFuture>() { @Override public void operationComplete(WriteFuture future) { IoSession session = future.getSession(); WsnSession wsnSession = (WsnSession)session; IoSession parentSession = wsnSession.getParent(); IoFilterChain parentFilterChain = parentSession.getFilterChain(); wsnConnector.removeBridgeFilters(parentFilterChain); try { boolean ok = waitForClientParentSessionCloseListenerEstabished.await(2, TimeUnit.SECONDS); if ( !ok ) { throw new RuntimeException("Failed to establish close listener on client-side parent session in time"); } } catch (InterruptedException e) { throw new RuntimeException("Failed to establish close listener on client-side parent session in time", e); } parentSession.close(true); } }); } }; wsnAcceptor.bind(address, acceptHandler, null); IoHandler connectHandler = new IoHandlerAdapter<IoSessionEx>() { @Override protected void doSessionOpened(IoSessionEx session) throws Exception { IoBufferAllocatorEx<?> allocator = session.getBufferAllocator(); session.write(allocator.wrap(wrap("Hello, world".getBytes()))); } }; ConnectFuture connectFuture = wsnConnector.connect(address, connectHandler, null); final WsnSession wsnSession = (WsnSession)connectFuture.await().getSession(); IoSession parentSession = wsnSession.getParent(); CloseFuture parentCloseFuture = parentSession.getCloseFuture(); parentCloseFuture.addListener(new IoFutureListener<CloseFuture>() { @Override public void operationComplete(CloseFuture future) { assert !wsnSession.isClosing(); IoBufferAllocatorEx<? extends WsBuffer> wsnAllocator = wsnSession.getBufferAllocator(); wsnSession.write(wsnAllocator.wrap(wrap("Goodbye, world".getBytes()))); } }); waitForClientParentSessionCloseListenerEstabished.countDown(); CloseFuture closeFuture = wsnSession.getCloseFuture(); assertTrue("WsnSession closed", closeFuture.await().isClosed()); }
@Override public void removeBridgeFilters(IoFilterChain filterChain) { removeFilter(filterChain, postUpgrade); removeFilter(filterChain, codec); removeFilter(filterChain, base64); removeFilter(filterChain, text); removeFilter(filterChain, WsAcceptor.CLOSE_FILTER); }
WsnConnector extends AbstractBridgeConnector<WsnSession> { @Override public void removeBridgeFilters(IoFilterChain filterChain) { removeFilter(filterChain, postUpgrade); removeFilter(filterChain, codec); removeFilter(filterChain, base64); removeFilter(filterChain, text); removeFilter(filterChain, WsAcceptor.CLOSE_FILTER); } }
WsnConnector extends AbstractBridgeConnector<WsnSession> { @Override public void removeBridgeFilters(IoFilterChain filterChain) { removeFilter(filterChain, postUpgrade); removeFilter(filterChain, codec); removeFilter(filterChain, base64); removeFilter(filterChain, text); removeFilter(filterChain, WsAcceptor.CLOSE_FILTER); } WsnConnector(); }
WsnConnector extends AbstractBridgeConnector<WsnSession> { @Override public void removeBridgeFilters(IoFilterChain filterChain) { removeFilter(filterChain, postUpgrade); removeFilter(filterChain, codec); removeFilter(filterChain, base64); removeFilter(filterChain, text); removeFilter(filterChain, WsAcceptor.CLOSE_FILTER); } WsnConnector(); @Resource(name = "bridgeServiceFactory") void setBridgeServiceFactory(BridgeServiceFactory bridgeServiceFactory); @Resource(name = "configuration") void setConfiguration(Properties configuration); @Resource(name = "schedulerProvider") void setSchedulerProvider(SchedulerProvider provider); @Resource(name = "resourceAddressFactory") void setResourceAddressFactory(ResourceAddressFactory factory); @Resource(name = "ws.connector") void setWsConnector(WsConnector connector); @Override TransportMetadata getTransportMetadata(); @Override void addBridgeFilters(IoFilterChain filterChain); @Override void removeBridgeFilters(IoFilterChain filterChain); }
WsnConnector extends AbstractBridgeConnector<WsnSession> { @Override public void removeBridgeFilters(IoFilterChain filterChain) { removeFilter(filterChain, postUpgrade); removeFilter(filterChain, codec); removeFilter(filterChain, base64); removeFilter(filterChain, text); removeFilter(filterChain, WsAcceptor.CLOSE_FILTER); } WsnConnector(); @Resource(name = "bridgeServiceFactory") void setBridgeServiceFactory(BridgeServiceFactory bridgeServiceFactory); @Resource(name = "configuration") void setConfiguration(Properties configuration); @Resource(name = "schedulerProvider") void setSchedulerProvider(SchedulerProvider provider); @Resource(name = "resourceAddressFactory") void setResourceAddressFactory(ResourceAddressFactory factory); @Resource(name = "ws.connector") void setWsConnector(WsConnector connector); @Override TransportMetadata getTransportMetadata(); @Override void addBridgeFilters(IoFilterChain filterChain); @Override void removeBridgeFilters(IoFilterChain filterChain);  }