@Test public void addFlow() throws Exception { AddFlow addFlow = AddFlow.builder() .flowId("some-flow-id") .encapsulationType(EncapsulationType.VLAN) .tunnelId(1001L) .direction(FlowDirection.REVERSE) .port(42) .build(); String switchId = "00:00:1b:45:18:d6:71:5a"; gate.listen(addFlow, switchId); CommandPacket commandPacket = getCommandPacket(); assertThat(commandPacket.getType()).isEqualTo(Type.ADD_FLOW); assertThat(commandPacket.getCommandCount()).isEqualTo(1); Any command = commandPacket.getCommand(0); assertThat(command.is(Control.AddFlow.class)).isTrue(); Control.AddFlow unpack = command.unpack(Control.AddFlow.class); Flow flow = unpack.getFlow(); assertThat(flow.getFlowId()).isEqualTo(addFlow.getFlowId()); assertThat(flow.getEncapsulationType().name()).isEqualTo(addFlow.getEncapsulationType().name()); assertThat(flow.getTunnelId()).isEqualTo(addFlow.getTunnelId()); assertThat(flow.getDirection()).isEqualTo(FlowDirection.toBoolean(addFlow.getDirection())); assertThat(flow.getUdpSrcPort()).isEqualTo(udpSrcPortOffset + addFlow.getPort()); assertThat(flow.getTransitEncapsulationType()).isEqualTo(Flow.EncapsulationType.VLAN); Map<Long, List<String>> vlanToSwitch = switchToVlanMapping.getVlan(); vlanToSwitch.forEach((vlan, switches) -> { if (switches.contains(switchId)) { assertThat(flow.getTransitTunnelId()).isEqualTo(vlan); } }); assertThat(flow.getDstMac()).isSubstringOf(switchId).isNotEqualTo(switchId); }
@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
    );   }