@Test public void testOpen() throws IOException, InterruptedException { final String offer1 = "relp_version=0"; final String offer2 = "relp_software=librelp,1.2.7,http: final String offer3 = "commands=syslog"; final String data = offer1 + "\n" + offer2 + "\n" + offer3; final RELPFrame openFrame = new RELPFrame.Builder() .txnr(1).command("open") .dataLength(data.length()) .data(data.getBytes(charset)) .build(); final String sender = "sender1"; final CapturingChannelResponder responder = new CapturingChannelResponder(); frameHandler.handle(openFrame, responder, sender); Assert.assertEquals(1, responder.responded); Assert.assertEquals(1, responder.responses.size()); final ChannelResponse response = responder.responses.get(0); final String responseData = new String(response.toByteArray(), charset); Assert.assertTrue(responseData.contains(offer1)); Assert.assertTrue(responseData.contains(offer2)); Assert.assertTrue(responseData.contains(offer3)); }
public void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { if (CMD_OPEN.equals(frame.getCommand())) { Map<String,String> offers = RELPResponse.parseOffers(frame.getData(), charset); ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.open(frame.getTxnr(), offers)); responder.addResponse(response); responder.respond(); } else if (CMD_CLOSE.equals(frame.getCommand())) { ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.ok(frame.getTxnr())); responder.addResponse(response); responder.respond(); dispatcher.completeConnection(key); } else { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(RELPMetadata.TXNR_KEY, String.valueOf(frame.getTxnr())); metadata.put(RELPMetadata.COMMAND_KEY, frame.getCommand()); final E event = eventFactory.create(frame.getData(), metadata, responder); events.offer(event); } }
RELPFrameHandler { public void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { if (CMD_OPEN.equals(frame.getCommand())) { Map<String,String> offers = RELPResponse.parseOffers(frame.getData(), charset); ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.open(frame.getTxnr(), offers)); responder.addResponse(response); responder.respond(); } else if (CMD_CLOSE.equals(frame.getCommand())) { ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.ok(frame.getTxnr())); responder.addResponse(response); responder.respond(); dispatcher.completeConnection(key); } else { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(RELPMetadata.TXNR_KEY, String.valueOf(frame.getTxnr())); metadata.put(RELPMetadata.COMMAND_KEY, frame.getCommand()); final E event = eventFactory.create(frame.getData(), metadata, responder); events.offer(event); } } }
RELPFrameHandler { public void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { if (CMD_OPEN.equals(frame.getCommand())) { Map<String,String> offers = RELPResponse.parseOffers(frame.getData(), charset); ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.open(frame.getTxnr(), offers)); responder.addResponse(response); responder.respond(); } else if (CMD_CLOSE.equals(frame.getCommand())) { ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.ok(frame.getTxnr())); responder.addResponse(response); responder.respond(); dispatcher.completeConnection(key); } else { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(RELPMetadata.TXNR_KEY, String.valueOf(frame.getTxnr())); metadata.put(RELPMetadata.COMMAND_KEY, frame.getCommand()); final E event = eventFactory.create(frame.getData(), metadata, responder); events.offer(event); } } RELPFrameHandler(final SelectionKey selectionKey,
                            final Charset charset,
                            final EventFactory<E> eventFactory,
                            final BlockingQueue<E> events,
                            final AsyncChannelDispatcher dispatcher,
                            final ComponentLog logger); }
RELPFrameHandler { public void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { if (CMD_OPEN.equals(frame.getCommand())) { Map<String,String> offers = RELPResponse.parseOffers(frame.getData(), charset); ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.open(frame.getTxnr(), offers)); responder.addResponse(response); responder.respond(); } else if (CMD_CLOSE.equals(frame.getCommand())) { ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.ok(frame.getTxnr())); responder.addResponse(response); responder.respond(); dispatcher.completeConnection(key); } else { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(RELPMetadata.TXNR_KEY, String.valueOf(frame.getTxnr())); metadata.put(RELPMetadata.COMMAND_KEY, frame.getCommand()); final E event = eventFactory.create(frame.getData(), metadata, responder); events.offer(event); } } RELPFrameHandler(final SelectionKey selectionKey,
                            final Charset charset,
                            final EventFactory<E> eventFactory,
                            final BlockingQueue<E> events,
                            final AsyncChannelDispatcher dispatcher,
                            final ComponentLog logger); void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender); }
RELPFrameHandler { public void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { if (CMD_OPEN.equals(frame.getCommand())) { Map<String,String> offers = RELPResponse.parseOffers(frame.getData(), charset); ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.open(frame.getTxnr(), offers)); responder.addResponse(response); responder.respond(); } else if (CMD_CLOSE.equals(frame.getCommand())) { ChannelResponse response = new RELPChannelResponse(encoder, RELPResponse.ok(frame.getTxnr())); responder.addResponse(response); responder.respond(); dispatcher.completeConnection(key); } else { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(RELPMetadata.TXNR_KEY, String.valueOf(frame.getTxnr())); metadata.put(RELPMetadata.COMMAND_KEY, frame.getCommand()); final E event = eventFactory.create(frame.getData(), metadata, responder); events.offer(event); } } RELPFrameHandler(final SelectionKey selectionKey,
                            final Charset charset,
                            final EventFactory<E> eventFactory,
                            final BlockingQueue<E> events,
                            final AsyncChannelDispatcher dispatcher,
                            final ComponentLog logger); void handle(final RELPFrame frame, final ChannelResponder<SocketChannel> responder, final String sender);  }