@Test public void clearFlowsTest() throws Exception { Headers headers = Headers.builder().correlationId("some-correlation-id").build(); ClearFlows clearFlows = ClearFlows.builder().headers(headers).build(); gate.listen(clearFlows); CommandPacket commandPacket = getCommandPacket(); assertThat(commandPacket.getType()).isEqualTo(Type.CLEAR_FLOWS); assertThat(commandPacket.getCommandList()).isEmpty(); }
@KafkaHandler void listen(@Payload AddFlow data, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String switchIdKey) { SwitchId switchId = new SwitchId(switchIdKey); Builder builder = CommandPacket.newBuilder(); Flow flow = Flow.newBuilder() .setFlowId(data.getFlowId()) .setEncapsulationType(EncapsulationType.forNumber(data.getEncapsulationType().ordinal())) .setTunnelId(data.getTunnelId()) .setTransitEncapsulationType(EncapsulationType.VLAN) .setTransitTunnelId(switchToVlanMap.get(switchIdKey)) .setDirection(FlowDirection.toBoolean(data.getDirection())) .setUdpSrcPort(udpSrcPortOffset + data.getPort()) .setDstMac(switchId.toMacAddress()) .build(); Control.AddFlow addFlow = Control.AddFlow.newBuilder().setFlow(flow).build(); builder.setType(Type.ADD_FLOW); builder.addCommand(Any.pack(addFlow)); CommandPacket packet = builder.build(); try { zeroMqClient.send(packet); } catch (InvalidProtocolBufferException e) { log.error("Marshalling error on {}", data); } }
Gate { @KafkaHandler void listen(@Payload AddFlow data, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String switchIdKey) { SwitchId switchId = new SwitchId(switchIdKey); Builder builder = CommandPacket.newBuilder(); Flow flow = Flow.newBuilder() .setFlowId(data.getFlowId()) .setEncapsulationType(EncapsulationType.forNumber(data.getEncapsulationType().ordinal())) .setTunnelId(data.getTunnelId()) .setTransitEncapsulationType(EncapsulationType.VLAN) .setTransitTunnelId(switchToVlanMap.get(switchIdKey)) .setDirection(FlowDirection.toBoolean(data.getDirection())) .setUdpSrcPort(udpSrcPortOffset + data.getPort()) .setDstMac(switchId.toMacAddress()) .build(); Control.AddFlow addFlow = Control.AddFlow.newBuilder().setFlow(flow).build(); builder.setType(Type.ADD_FLOW); builder.addCommand(Any.pack(addFlow)); CommandPacket packet = builder.build(); try { zeroMqClient.send(packet); } catch (InvalidProtocolBufferException e) { log.error("Marshalling error on {}", data); } } }
Gate { @KafkaHandler void listen(@Payload AddFlow data, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String switchIdKey) { SwitchId switchId = new SwitchId(switchIdKey); Builder builder = CommandPacket.newBuilder(); Flow flow = Flow.newBuilder() .setFlowId(data.getFlowId()) .setEncapsulationType(EncapsulationType.forNumber(data.getEncapsulationType().ordinal())) .setTunnelId(data.getTunnelId()) .setTransitEncapsulationType(EncapsulationType.VLAN) .setTransitTunnelId(switchToVlanMap.get(switchIdKey)) .setDirection(FlowDirection.toBoolean(data.getDirection())) .setUdpSrcPort(udpSrcPortOffset + data.getPort()) .setDstMac(switchId.toMacAddress()) .build(); Control.AddFlow addFlow = Control.AddFlow.newBuilder().setFlow(flow).build(); builder.setType(Type.ADD_FLOW); builder.addCommand(Any.pack(addFlow)); CommandPacket packet = builder.build(); try { zeroMqClient.send(packet); } catch (InvalidProtocolBufferException e) { log.error("Marshalling error on {}", data); } } Gate(@Autowired KafkaTemplate<String, Object> template,
                @Autowired ZeroMqClient zeroMqClient,
                @Autowired SwitchToVlanMapping switchToVlanMapping
    ); }
Gate { @KafkaHandler void listen(@Payload AddFlow data, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String switchIdKey) { SwitchId switchId = new SwitchId(switchIdKey); Builder builder = CommandPacket.newBuilder(); Flow flow = Flow.newBuilder() .setFlowId(data.getFlowId()) .setEncapsulationType(EncapsulationType.forNumber(data.getEncapsulationType().ordinal())) .setTunnelId(data.getTunnelId()) .setTransitEncapsulationType(EncapsulationType.VLAN) .setTransitTunnelId(switchToVlanMap.get(switchIdKey)) .setDirection(FlowDirection.toBoolean(data.getDirection())) .setUdpSrcPort(udpSrcPortOffset + data.getPort()) .setDstMac(switchId.toMacAddress()) .build(); Control.AddFlow addFlow = Control.AddFlow.newBuilder().setFlow(flow).build(); builder.setType(Type.ADD_FLOW); builder.addCommand(Any.pack(addFlow)); CommandPacket packet = builder.build(); try { zeroMqClient.send(packet); } catch (InvalidProtocolBufferException e) { log.error("Marshalling error on {}", data); } } Gate(@Autowired KafkaTemplate<String, Object> template,
                @Autowired ZeroMqClient zeroMqClient,
                @Autowired SwitchToVlanMapping switchToVlanMapping
    );  }
Gate { @KafkaHandler void listen(@Payload AddFlow data, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String switchIdKey) { SwitchId switchId = new SwitchId(switchIdKey); Builder builder = CommandPacket.newBuilder(); Flow flow = Flow.newBuilder() .setFlowId(data.getFlowId()) .setEncapsulationType(EncapsulationType.forNumber(data.getEncapsulationType().ordinal())) .setTunnelId(data.getTunnelId()) .setTransitEncapsulationType(EncapsulationType.VLAN) .setTransitTunnelId(switchToVlanMap.get(switchIdKey)) .setDirection(FlowDirection.toBoolean(data.getDirection())) .setUdpSrcPort(udpSrcPortOffset + data.getPort()) .setDstMac(switchId.toMacAddress()) .build(); Control.AddFlow addFlow = Control.AddFlow.newBuilder().setFlow(flow).build(); builder.setType(Type.ADD_FLOW); builder.addCommand(Any.pack(addFlow)); CommandPacket packet = builder.build(); try { zeroMqClient.send(packet); } catch (InvalidProtocolBufferException e) { log.error("Marshalling error on {}", data); } } Gate(@Autowired KafkaTemplate<String, Object> template,
                @Autowired ZeroMqClient zeroMqClient,
                @Autowired SwitchToVlanMapping switchToVlanMapping
    );   }