@Test public void testJson() throws IOException, InterruptedException { final byte jsonPayload[] = new byte[]{ 0x7b, 0x22, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x3a, 0x20, 0x22, 0x74, 0x65, 0x73, 0x74, 0x2d, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x22, 0x2c, 0x20, 0x22, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x3a, 0x20, 0x22, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x7d }; final BeatsFrame jsonFrame = new BeatsFrame.Builder() .version((byte) 0x32) .frameType((byte) 0x4a) .seqNumber(1) .dataSize(45) .payload(jsonPayload) .build(); final String sender = "sender1"; final CapturingChannelResponder responder = new CapturingChannelResponder(); frameHandler.handle(jsonFrame, responder, sender); Assert.assertEquals(0, responder.responded); Assert.assertEquals(1, events.size()); final BeatsEvent event = events.poll(); Assert.assertEquals("{\"message\": \"test-content\", \"field\": \"value\"}", new String(event.getData(), charset)); }
public void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(BeatsMetadata.SEQNUMBER_KEY, String.valueOf(frame.getSeqNumber())); String line = ""; if (frame.getFrameType() == FRAME_JSON ) { final E event = eventFactory.create(frame.getPayload(), metadata, responder); events.offer(event); } }
BeatsFrameHandler { public void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(BeatsMetadata.SEQNUMBER_KEY, String.valueOf(frame.getSeqNumber())); String line = ""; if (frame.getFrameType() == FRAME_JSON ) { final E event = eventFactory.create(frame.getPayload(), metadata, responder); events.offer(event); } } }
BeatsFrameHandler { public void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(BeatsMetadata.SEQNUMBER_KEY, String.valueOf(frame.getSeqNumber())); String line = ""; if (frame.getFrameType() == FRAME_JSON ) { final E event = eventFactory.create(frame.getPayload(), metadata, responder); events.offer(event); } } BeatsFrameHandler(final SelectionKey selectionKey,
                             final Charset charset,
                             final EventFactory<E> eventFactory,
                             final BlockingQueue<E> events,
                             final AsyncChannelDispatcher dispatcher,
                             final ComponentLog logger); }
BeatsFrameHandler { public void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(BeatsMetadata.SEQNUMBER_KEY, String.valueOf(frame.getSeqNumber())); String line = ""; if (frame.getFrameType() == FRAME_JSON ) { final E event = eventFactory.create(frame.getPayload(), metadata, responder); events.offer(event); } } BeatsFrameHandler(final SelectionKey selectionKey,
                             final Charset charset,
                             final EventFactory<E> eventFactory,
                             final BlockingQueue<E> events,
                             final AsyncChannelDispatcher dispatcher,
                             final ComponentLog logger); void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender); }
BeatsFrameHandler { public void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender) throws IOException, InterruptedException { final Map<String, String> metadata = EventFactoryUtil.createMapWithSender(sender.toString()); metadata.put(BeatsMetadata.SEQNUMBER_KEY, String.valueOf(frame.getSeqNumber())); String line = ""; if (frame.getFrameType() == FRAME_JSON ) { final E event = eventFactory.create(frame.getPayload(), metadata, responder); events.offer(event); } } BeatsFrameHandler(final SelectionKey selectionKey,
                             final Charset charset,
                             final EventFactory<E> eventFactory,
                             final BlockingQueue<E> events,
                             final AsyncChannelDispatcher dispatcher,
                             final ComponentLog logger); void handle(final BeatsFrame frame, final ChannelResponder<SocketChannel> responder, final String sender); static final byte FRAME_WINDOWSIZE; }