@Test public void sendStatsTest() throws Exception { Builder bucketBuilder = FlowLatencyPacketBucket.newBuilder(); FlowLatencyPacket packet1 = FlowLatencyPacket.newBuilder() .setFlowId("some-flow-id-1") .setDirection(false) .setT0(100) .setT1(150) .setPacketId(1).build(); FlowLatencyPacket packet2 = FlowLatencyPacket.newBuilder() .setDirection(true) .setT0(200) .setT1(250) .setPacketId(2).build(); bucketBuilder.addPacket(packet1); bucketBuilder.addPacket(packet2); statsCollector.sendStats(bucketBuilder.build()); ArgumentCaptor<InfoMessage> argument = ArgumentCaptor.forClass(InfoMessage.class); verify(template).send(eq(toStorm), eq(packet1.getFlowId()), argument.capture()); InfoMessage packet1Message = argument.getValue(); FlowRttStatsData statsPacket1 = (FlowRttStatsData) packet1Message.getData(); assertThat(statsPacket1).extracting( FlowRttStatsData::getFlowId, FlowRttStatsData::getT0, FlowRttStatsData::getT1) .contains(packet1.getFlowId(), packet1.getT0(), packet1.getT1()); assertThat(statsPacket1) .extracting(FlowRttStatsData::getDirection) .isEqualTo("forward"); verify(template).send(eq(toStorm), eq(packet2.getFlowId()), argument.capture()); InfoMessage packet2Message = argument.getValue(); FlowRttStatsData statsPacket2 = (FlowRttStatsData) packet2Message.getData(); assertThat(statsPacket2).extracting( FlowRttStatsData::getFlowId, FlowRttStatsData::getT0, FlowRttStatsData::getT1) .contains(packet2.getFlowId(), packet2.getT0(), packet2.getT1()); assertThat(statsPacket2) .extracting(FlowRttStatsData::getDirection) .isEqualTo("reverse"); }
void sendStats(FlowLatencyPacketBucket flowLatencyPacketBucket) throws InvalidProtocolBufferException { long currentTimeMillis = System.currentTimeMillis(); for (FlowLatencyPacket packet : flowLatencyPacketBucket.getPacketList()) { FlowRttStatsData data = new FlowRttStatsData( packet.getFlowId(), FlowDirection.fromBoolean(packet.getDirection()).name().toLowerCase(), packet.getT0(), packet.getT1() ); InfoMessage message = new InfoMessage(data, currentTimeMillis, String.format("stats42-%s-%d", sessionId, packet.getPacketId())); log.debug("InfoMessage {}", message); template.send(toStorm, packet.getFlowId(), message); } }
StatsCollector extends Thread { void sendStats(FlowLatencyPacketBucket flowLatencyPacketBucket) throws InvalidProtocolBufferException { long currentTimeMillis = System.currentTimeMillis(); for (FlowLatencyPacket packet : flowLatencyPacketBucket.getPacketList()) { FlowRttStatsData data = new FlowRttStatsData( packet.getFlowId(), FlowDirection.fromBoolean(packet.getDirection()).name().toLowerCase(), packet.getT0(), packet.getT1() ); InfoMessage message = new InfoMessage(data, currentTimeMillis, String.format("stats42-%s-%d", sessionId, packet.getPacketId())); log.debug("InfoMessage {}", message); template.send(toStorm, packet.getFlowId(), message); } } }
StatsCollector extends Thread { void sendStats(FlowLatencyPacketBucket flowLatencyPacketBucket) throws InvalidProtocolBufferException { long currentTimeMillis = System.currentTimeMillis(); for (FlowLatencyPacket packet : flowLatencyPacketBucket.getPacketList()) { FlowRttStatsData data = new FlowRttStatsData( packet.getFlowId(), FlowDirection.fromBoolean(packet.getDirection()).name().toLowerCase(), packet.getT0(), packet.getT1() ); InfoMessage message = new InfoMessage(data, currentTimeMillis, String.format("stats42-%s-%d", sessionId, packet.getPacketId())); log.debug("InfoMessage {}", message); template.send(toStorm, packet.getFlowId(), message); } } StatsCollector(KafkaTemplate<String, Object> template); }
StatsCollector extends Thread { void sendStats(FlowLatencyPacketBucket flowLatencyPacketBucket) throws InvalidProtocolBufferException { long currentTimeMillis = System.currentTimeMillis(); for (FlowLatencyPacket packet : flowLatencyPacketBucket.getPacketList()) { FlowRttStatsData data = new FlowRttStatsData( packet.getFlowId(), FlowDirection.fromBoolean(packet.getDirection()).name().toLowerCase(), packet.getT0(), packet.getT1() ); InfoMessage message = new InfoMessage(data, currentTimeMillis, String.format("stats42-%s-%d", sessionId, packet.getPacketId())); log.debug("InfoMessage {}", message); template.send(toStorm, packet.getFlowId(), message); } } StatsCollector(KafkaTemplate<String, Object> template); @Override void run(); }
StatsCollector extends Thread { void sendStats(FlowLatencyPacketBucket flowLatencyPacketBucket) throws InvalidProtocolBufferException { long currentTimeMillis = System.currentTimeMillis(); for (FlowLatencyPacket packet : flowLatencyPacketBucket.getPacketList()) { FlowRttStatsData data = new FlowRttStatsData( packet.getFlowId(), FlowDirection.fromBoolean(packet.getDirection()).name().toLowerCase(), packet.getT0(), packet.getT1() ); InfoMessage message = new InfoMessage(data, currentTimeMillis, String.format("stats42-%s-%d", sessionId, packet.getPacketId())); log.debug("InfoMessage {}", message); template.send(toStorm, packet.getFlowId(), message); } } StatsCollector(KafkaTemplate<String, Object> template); @Override void run();  }