Java 类org.projectfloodlight.openflow.protocol.OFMessage 实例源码

/**
 * Verify that the given exception event capture (as returned by
 * getAndInitExceptionCapture) has thrown an exception of the given
 * expectedExceptionClass.
 * Resets the capture
 * @param err
 */
void verifyExceptionCaptured(
        OFMessage err, Class<? extends Throwable> expectedExceptionClass) {

    Throwable caughtEx = null;
    // This should purposely cause an exception
    try{
        switchHandler.processOFMessage(err);
    }
    catch(Exception e){
        // Capture the exception
        caughtEx = e;
    }

    assertThat(caughtEx, CoreMatchers.instanceOf(expectedExceptionClass));
}

public void dispatchMessage(IOFSwitch sw, OFMessage msg, FloodlightContext bc) {
    List<IOFMessageListener> theListeners = listeners.get(msg.getType()).getOrderedListeners();
    if (theListeners != null) {
        Command result = Command.CONTINUE;
        Iterator<IOFMessageListener> it = theListeners.iterator();
        if (OFType.PACKET_IN.equals(msg.getType())) {
            OFPacketIn pi = (OFPacketIn)msg;
            Ethernet eth = new Ethernet();
            eth.deserialize(pi.getData(), 0, pi.getData().length);
            IFloodlightProviderService.bcStore.put(bc,
                    IFloodlightProviderService.CONTEXT_PI_PAYLOAD,
                    eth);
        }
        while (it.hasNext() && !Command.STOP.equals(result)) {
            result = it.next().receive(sw, msg, bc);
        }
    }
}

/**
 * Send the configuration requests to tell the switch we want full
 * packets
 * @throws IOException
 */
private void sendHandshakeSetConfig() {
    // Ensure we receive the full packet via PacketIn
    // FIXME: We don't set the reassembly flags.
    OFSetConfig configSet = factory.buildSetConfig()
            .setXid(handshakeTransactionIds--)
            .setMissSendLen(0xffff)
            .build();

    // Barrier
    OFBarrierRequest barrier = factory.buildBarrierRequest()
            .setXid(handshakeTransactionIds--)
            .build();

    // Verify (need barrier?)
    OFGetConfigRequest configReq = factory.buildGetConfigRequest()
            .setXid(handshakeTransactionIds--)
            .build();
    List<OFMessage> msgList = ImmutableList.<OFMessage>of(configSet, barrier, configReq);
    mainConnection.write(msgList);
}

@Override
public net.floodlightcontroller.core.IListener.Command receive(
        IOFSwitch sw, OFMessage msg, FloodlightContext cntx) {
    System.out.println("flow expired: "+sw.toString() + msg.toString());

    //OFFlowRemoved flowRemoved = (OFFlowRemoved) msg;

    if (!switchStates.containsKey(sw))
        switchStates.put(sw, new ObfuscationSwitchState(sw));

    if (msg.getType() == OFType.FLOW_REMOVED) {
        OFFlowRemoved flowRemoved = (OFFlowRemoved) msg;
        System.out.println("flow expired: "+sw.toString() + "dst: " + flowRemoved.getCookie());
        long dst = flowRemoved.getCookie().getValue();
        ObfuscationHeader oHeader = new ObfuscationHeader();
        Match match = flowRemoved.getMatch();

        switchStates.get(sw).removeDestinationID(dst);

    }

    return Command.CONTINUE;
}

private OFMessage createHubPacketOut(IOFSwitch sw, OFMessage msg) {
    OFPacketIn pi = (OFPacketIn) msg;
    OFPacketOut.Builder pob = sw.getOFFactory().buildPacketOut();
    pob.setBufferId(pi.getBufferId()).setXid(pi.getXid()).setInPort((pi.getVersion().compareTo(OFVersion.OF_12) < 0 ? pi.getInPort() : pi.getMatch().get(MatchField.IN_PORT)));

    // set actions
    OFActionOutput.Builder actionBuilder = sw.getOFFactory().actions().buildOutput();
    actionBuilder.setPort(OFPort.FLOOD);
    pob.setActions(Collections.singletonList((OFAction) actionBuilder.build()));

    // set data if it is included in the packetin
    if (pi.getBufferId() == OFBufferId.NO_BUFFER) {
        byte[] packetData = pi.getData();
        pob.setData(packetData);
    }
    return pob.build();  
}

@Test
public void moveToWaitHello() throws Exception {
    resetChannel();
    channel.write(capture(writeCapture));
    expectLastCall().andReturn(null).once();
    replay(channel);
    // replay unused mocks
    replay(messageEvent);

    handler.channelConnected(ctx, channelStateEvent);

    List<OFMessage> msgs = getMessagesFromCapture();
    assertEquals(1, msgs.size());
    assertEquals(OFType.HELLO, msgs.get(0).getType());
    assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitHelloState.class));
    verifyUniqueXids(msgs);
}

@Override
public void handleOutgoingMessage(IOFSwitch sw, OFMessage m) {
    FloodlightContext bc = new FloodlightContext();
    List<IOFMessageListener> msgListeners = null;
    if (listeners.containsKey(m.getType())) {
        msgListeners = listeners.get(m.getType()).getOrderedListeners();
    }

    if (msgListeners != null) {
        for (IOFMessageListener listener : msgListeners) {
            if (Command.STOP.equals(listener.receive(sw, m, bc))) {
                break;
            }
        }
    }
}

protected void doWrite(boolean expectWrite, 
                       OFMessageDamperMockSwitch sw, 
                       OFMessage msg) throws IOException {

    boolean result;
    sw.reset();
    result = damper.write(sw, msg);

    if (expectWrite) {
        assertEquals(true, result);
        sw.assertMessageWasWritten(msg);
    } else {
        assertEquals(false, result);
        sw.assertNoMessageWritten();
    }
}

/** Move the channel from scratch to WAIT_FEATURES_REPLY state
 * Builds on moveToWaitHello()
 * adds testing for WAIT_HELLO state
 */
@Test
public void moveToWaitFeaturesReply() throws Exception {

    moveToWaitHello();

    resetChannel();
    expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).atLeastOnce();
    replay(channel);

    OFMessage hello = factory.buildHello().build();
    sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(hello));

    List<OFMessage> msgs = getMessagesFromCapture();
    assertEquals(1, msgs.size());
    assertEquals(OFType.FEATURES_REQUEST, msgs.get(0).getType());
    verifyUniqueXids(msgs);

    assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitFeaturesReplyState.class));
}

@Override
protected Object encode(ChannelHandlerContext ctx, Channel channel,
                        Object msg) throws Exception {
    if (!(msg instanceof List)) {
        return msg;
    }

    @SuppressWarnings("unchecked")
    List<OFMessage> msglist = (List<OFMessage>) msg;
    /* XXX S can't get length of OFMessage in loxigen's openflowj??
    int size = 0;
    for (OFMessage ofm : msglist) {
        size += ofm.getLengthU();
    }*/

    ChannelBuffer buf = ChannelBuffers.dynamicBuffer();

    for (OFMessage ofm : msglist) {
        if (ofm != null) {
            ofm.writeTo(buf);
        }
    }
    return buf;
}

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
        throws Exception {
    if (e.getMessage() instanceof List) {
        @SuppressWarnings("unchecked")
        List<OFMessage> msglist = (List<OFMessage>) e.getMessage();


        for (OFMessage ofm : msglist) {
            // Do the actual packet processing
            state.processOFMessage(this, ofm);
        }
    } else {
        state.processOFMessage(this, (OFMessage) e.getMessage());
    }
}

@Override
public void handleOutgoingMessage(IOFSwitch sw, OFMessage m) {
    if (sw == null)
        throw new NullPointerException("Switch must not be null");
    if (m == null)
        throw new NullPointerException("OFMessage must not be null");

    // FIXME floodlight context not supported any more
    FloodlightContext bc = new FloodlightContext();

    List<IOFMessageListener> listeners = null;
    if (messageListeners.containsKey(m.getType())) {
        listeners = messageListeners.get(m.getType()).getOrderedListeners();
    }

    if (listeners != null) {
        for (IOFMessageListener listener : listeners) {
            if (Command.STOP.equals(listener.receive(sw, m, bc))) {
                break;
            }
        }
    }
}

/** write a packetOut, which is buffered */
@Test(timeout = 5000)
public void testSingleMessageWrite() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFPacketOut packetOut = factory.buildPacketOut()
            .setData(new byte[] { 0x01, 0x02, 0x03, 0x04 })
            .setActions(ImmutableList.<OFAction>of( factory.actions().output(OFPort.of(1), 0)))
            .build();

    conn.write(packetOut);
    assertThat("Write should have been flushed", cMsgList.hasCaptured(), equalTo(true));

    List<OFMessage> value = cMsgList.getValue();
    logger.info("Captured channel write: "+value);
    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(packetOut));
}

/** Move the channel from scratch to WAIT_FEATURES_REPLY state
 * Builds on moveToWaitHello()
 * adds testing for WAIT_HELLO state
 */
@Test
public void moveToWaitFeaturesReply() throws Exception {

    moveToWaitHello();

    resetChannel();
    channel.write(capture(writeCapture));
    expectLastCall().andReturn(null).atLeastOnce();
    replay(channel);

    OFMessage hello = factory.buildHello().build();
    sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(hello));

    List<OFMessage> msgs = getMessagesFromCapture();
    assertEquals(1, msgs.size());
    assertEquals(OFType.FEATURES_REQUEST, msgs.get(0).getType());
    verifyUniqueXids(msgs);

    assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitFeaturesReplyState.class));
}

public boolean handleGenericDeliverable(OFMessage reply) {
    counters.updateReadStats(reply);
    @SuppressWarnings("unchecked")
    Deliverable<OFMessage> deliverable =
    (Deliverable<OFMessage>) this.xidDeliverableMap.get(reply.getXid());
    if (deliverable != null) {
        if(reply instanceof OFErrorMsg) {
            deliverable.deliverError(new OFErrorMsgException((OFErrorMsg) reply));
        } else {
            deliverable.deliver(reply);
        }
        if (deliverable.isDone())
            this.xidDeliverableMap.remove(reply.getXid());
        return true;
    } else {
        return false;
    }
}

/**
 * Writes a list of OFMessages to a switch
 * @param dpid The datapath ID of the switch to write to
 * @param messages The list of OFMessages to write.
 */
@LogMessageDoc(level="ERROR",
        message="Tried to write to switch {switch} but got {error}",
        explanation="An I/O error occured while trying to write a " +
                "static flow to a switch",
                recommendation=LogMessageDoc.CHECK_SWITCH)
private void writeOFMessagesToSwitch(DatapathId dpid, List<OFMessage> messages) {
    IOFSwitch ofswitch = switchService.getSwitch(dpid);
    if (ofswitch != null) {  // is the switch connected
        if (log.isDebugEnabled()) {
            log.debug("Sending {} new entries to {}", messages.size(), dpid);
        }
        ofswitch.write(messages);
        ofswitch.flush();
    }
}

private void removeRule(FlowRule flowRule) {
    Dpid dpid = Dpid.dpid(flowRule.deviceId().uri());
    OpenFlowSwitch sw = controller.getSwitch(dpid);

    if (sw == null) {
        return;
    }

    FlowRuleExtPayLoad flowRuleExtPayLoad = flowRule.payLoad();
    if (hasPayload(flowRuleExtPayLoad)) {
        OFMessage msg = new ThirdPartyMessage(flowRuleExtPayLoad.payLoad());
        sw.sendMsg(msg);
        return;
    }
    sw.sendMsg(FlowModBuilder.builder(flowRule, sw.factory(),
                                      Optional.empty(), Optional.of(driverService)).buildFlowDel());

    if (adaptiveFlowSampling) {
        // Remove TypedFlowEntry to deviceFlowEntries in NewAdaptiveFlowStatsCollector
        NewAdaptiveFlowStatsCollector collector = afsCollectors.get(dpid);
        if (collector != null) {
            collector.removeFlows(flowRule);
        }
    }
}

@Override
public void run() {
    for (OFMessage m : msglist) {
        if (logger.isTraceEnabled())
            logger.trace("{}: send {}", this, m);
        counters.updateWriteStats(m);
    }
    channel.writeAndFlush(msglist);
}

@Override
protected OFMessage getRoleReply(long xid, OFControllerRole role) {
    OFNiciraControllerRoleReply roleReply = factory.buildNiciraControllerRoleReply()
            .setXid(xid)
            .setRole(NiciraRoleUtils.ofRoleToNiciraRole(role))
            .build();
    return roleReply;
}

@Override
public <R extends OFMessage> ListenableFuture<R>
        writeRequest(OFRequest<R> request) {
    RequestAndFuture<R> raf = new RequestAndFuture<>(request);
    messages.add(request);
    requests.put(request.getXid(), raf);
    return raf.getReplyFuture();
}

/**
 * Move the channel from scratch to WAIT_INITIAL_ROLE state via
 * WAIT_SWITCH_DRIVER_SUB_HANDSHAKE
 * Does extensive testing for the WAIT_SWITCH_DRIVER_SUB_HANDSHAKE state
 *
 */
@Test
public void testSwitchDriverSubHandshake()
        throws Exception {
    moveToWaitSwitchDriverSubHandshake();

    //-------------------------------------------------
    //-------------------------------------------------
    // Send a message to the handler, it should be passed to the
    // switch's sub-handshake handling. After this message the
    // sub-handshake will be complete
    // FIXME:LOJI: With Andi's fix for a default Match object we won't
    // need to build/set this match object

    Match match = factory.buildMatch().build();
    OFMessage m = factory.buildFlowRemoved().setMatch(match).build();
    resetToStrict(sw);
    sw.processDriverHandshakeMessage(m);
    expectLastCall().once();
    expect(sw.isDriverHandshakeComplete()).andReturn(true).once();
    replay(sw);

    switchHandler.processOFMessage(m);

    assertThat(switchHandler.getStateForTesting(),
            CoreMatchers.instanceOf(OFSwitchHandshakeHandler.WaitAppHandshakeState.class));
    assertThat("Unexpected message captured", connection.getMessages(), Matchers.empty());
    verify(sw);
}

@Override
public net.floodlightcontroller.core.IListener.Command
        receive(IOFSwitch sw, OFMessage msg, FloodlightContext cntx) {
    switch (msg.getType()) {
        case PACKET_IN:
            return processPacketIn(sw, (OFPacketIn)msg, cntx);
        default:
            break;
    }
    log.warn("Received unexpected message {}", msg);
    return Command.CONTINUE;
}

/**
 * Extract the list of OFMessages that was captured by the Channel.write()
 * capture. Will check that something was actually captured first. We'll
 * collapse the messages from multiple writes into a single list of
 * OFMessages.
 * Resets the channelWriteCapture.
 */
List<OFMessage> getMessagesFromCapture() {
    List<OFMessage> msgs = new ArrayList<OFMessage>();

    assertTrue("No write on channel was captured",
               writeCapture.hasCaptured());
    List<List<OFMessage>> capturedVals = writeCapture.getValues();

    for (List<OFMessage> oneWriteList: capturedVals)
        msgs.addAll(oneWriteList);
    writeCapture.reset();
    return msgs;
}

@Override
public void channelRead0(ChannelHandlerContext ctx, Iterable<OFMessage> msgList) throws Exception {
    for (OFMessage ofm : msgList) {
        try {
            // Do the actual packet processing
            state.processOFMessage(ofm);
        }
        catch (Exception ex) {
            // We are the last handler in the stream, so run the
            // exception through the channel again by passing in
            // ctx.getChannel().
            ctx.fireExceptionCaught(ex);
        }
    }
}

private OFMessage createHubFlowMod(IOFSwitch sw, OFMessage msg) {
    OFPacketIn pi = (OFPacketIn) msg;
    OFFlowAdd.Builder fmb = sw.getOFFactory().buildFlowAdd();
    fmb.setBufferId(pi.getBufferId())
    .setXid(pi.getXid());

    // set actions
    OFActionOutput.Builder actionBuilder = sw.getOFFactory().actions().buildOutput();
    actionBuilder.setPort(OFPort.FLOOD);
    fmb.setActions(Collections.singletonList((OFAction) actionBuilder.build()));

    return fmb.build();
}

private void sendMsgsOnChannel(List<OFMessage> msgs) {
    if (channel.isConnected()) {
        channel.write(msgs);
        agent.processDownstreamMessage(dpid, msgs);
    } else {
        log.warn("Dropping messages for switch {} because channel is not connected: {}",
                 dpid, msgs);
    }
}

/** Move the channel from scratch to SLAVE state
 * Builds on doMoveToWaitInitialRole()
 * adds testing for WAIT_INITAL_ROLE state
 *
 * We let the initial role request time out. The switch should be
 * disconnected
 */
@Test
public void testInitialMoveToSlaveTimeout() throws Exception {
    int timeout = 50;
    switchHandler.useRoleChangerWithOtherTimeoutForTesting(timeout);

    // first, move us to WAIT_INITIAL_ROLE_STATE
    moveToWaitInitialRole();
    assertThat(switchHandler.getStateForTesting(), CoreMatchers.instanceOf(OFSwitchHandshakeHandler.WaitInitialRoleState.class));

    // Set the role
    setupSwitchSendRoleRequestAndVerify(null, OFControllerRole.ROLE_SLAVE);
    assertThat(switchHandler.getStateForTesting(), CoreMatchers.instanceOf(OFSwitchHandshakeHandler.WaitInitialRoleState.class));

    // prepare mocks and inject the role reply message
    reset(sw);
    sw.setAttribute(IOFSwitch.SWITCH_SUPPORTS_NX_ROLE, false);
    expectLastCall().once();
    sw.setControllerRole(OFControllerRole.ROLE_SLAVE);
    expectLastCall().once();
    sw.disconnect(); // Make sure we disconnect
    expectLastCall().once();
    replay(sw);

    // Apparently this can be any type of message for this test?!
    OFMessage m = factory.buildBarrierReply().build();

    Thread.sleep(timeout+5);
    switchHandler.processOFMessage(m);
}

@Override
void processOFMessage(OFMessage m) {
    if(m.getType() == OFType.PORT_STATUS){
        OFPortStatus status = (OFPortStatus) m;
        handlePortStatusMessage(status, false);
    }
    else if(plugin != null){
        this.plugin.processOFMessage(m);
    }
    else{
        super.processOFMessage(m);
    }
}

@Test(timeout = 5000)
public void testWriteRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFEchoRequest echoRequest = factory.echoRequest(new byte[] {});
    ListenableFuture<OFEchoReply> future = conn.writeRequest(echoRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

    eventLoop.runTasks();
    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(echoRequest));

    assertThat("Future should not be complete yet", future.isDone(), equalTo(false));

    OFEchoReply echoReply = factory.buildEchoReply()
            .setXid(echoRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(echoReply),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));
    assertThat(future.get(), equalTo(echoReply));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/**
 * Process an OF message received on the channel and
 * update state accordingly.
 *
 * The main "event" of the state machine. Process the received message,
 * send follow up message if required and update state if required.
 *
 * Switches on the message type and calls more specific event handlers
 * for each individual OF message type. If we receive a message that
 * is supposed to be sent from a controller to a switch we throw
 * a SwitchStateExeption.
 *
 * The more specific handlers can also throw SwitchStateExceptions
 *
 * @param h The OFChannelHandler that received the message
 * @param m The message we received.
 * @throws SwitchStateException
 * @throws IOException
 */
void processOFMessage(OFMessage m) {
    roleChanger.checkTimeout();
    switch(m.getType()) {
    case BARRIER_REPLY:
        processOFBarrierReply((OFBarrierReply) m);
        break;
    case ERROR:
        processOFError((OFErrorMsg) m);
        break;
    case FLOW_REMOVED:
        processOFFlowRemoved((OFFlowRemoved) m);
        break;
    case GET_CONFIG_REPLY:
        processOFGetConfigReply((OFGetConfigReply) m);
        break;
    case PACKET_IN:
        processOFPacketIn((OFPacketIn) m);
        break;
    case PORT_STATUS:
        processOFPortStatus((OFPortStatus) m);
        break;
    case QUEUE_GET_CONFIG_REPLY:
        processOFQueueGetConfigReply((OFQueueGetConfigReply) m);
        break;
    case STATS_REPLY:
        processOFStatsReply((OFStatsReply) m);
        break;
    case ROLE_REPLY:
        processOFRoleReply((OFRoleReply) m);
        break;
    case EXPERIMENTER:
        processOFExperimenter((OFExperimenter) m);
        break;
    default:
        illegalMessageReceived(m);
        break;
    }
}

@Override
public void recordEndTimePktIn(IOFSwitch sw, OFMessage m, FloodlightContext cntx) {
    if (isEnabled()) {
        long procTimeNs = System.nanoTime() - startTimePktNs;
        ctb.updatePerPacketCounters(procTimeNs);

        if (ptWarningThresholdInNano > 0 && 
                procTimeNs > ptWarningThresholdInNano) {
            logger.warn("Time to process packet-in exceeded threshold: {}", 
                        procTimeNs/1000);
        }
    }
}

@Test
public void testMasterSlaveWrites() {
    OFFactory factory = OFFactories.getFactory(OFVersion.OF_13);
    OFFlowAdd fa = factory.buildFlowAdd().build();
    OFFlowStatsRequest fsr = factory.buildFlowStatsRequest().build();
    List<OFMessage> msgList = new ArrayList<OFMessage>();
    msgList.add(fa);
    msgList.add(fsr);

    reset(switchManager);
       expect(switchManager.isCategoryRegistered(LogicalOFMessageCategory.MAIN)).andReturn(true).times(6);
       switchManager.handleOutgoingMessage(sw, fa);
       expectLastCall().times(2);
       switchManager.handleOutgoingMessage(sw, fsr);
       expectLastCall().times(4);
       replay(switchManager);

    /* test master -- both messages should be written */
    sw.setControllerRole(OFControllerRole.ROLE_MASTER);
    assertTrue(sw.write(fa));
    assertTrue(sw.write(fsr));
    assertEquals(Collections.<OFMessage>emptyList(), sw.write(msgList));

    /* test slave -- flow-add (mod op) should fail each time; flow stats (read op) should pass */
    sw.setControllerRole(OFControllerRole.ROLE_SLAVE);
    assertFalse(sw.write(fa)); /* flow-add should be stopped (mod op) */
    assertTrue(sw.write(fsr)); /* stats request makes it (read op) */
    assertEquals(Collections.<OFMessage>singletonList(fa), sw.write(msgList)); /* return bad flow-add */
}

/** Start from scratch and reply with an unexpected error to the role
 * change request
 * Builds on doMoveToWaitInitialRole()
 * adds testing for WAIT_INITAL_ROLE state
 */
@Test
public void testInitialRoleChangeOtherError() throws Exception {
    // first, move us to WAIT_INITIAL_ROLE_STATE
    moveToWaitInitialRole();
    assertThat(switchHandler.getStateForTesting(), CoreMatchers.instanceOf(OFSwitchHandshakeHandler.WaitInitialRoleState.class));

    // Set the role
    long xid = setupSwitchSendRoleRequestAndVerify(null, OFControllerRole.ROLE_MASTER);
    assertThat(switchHandler.getStateForTesting(), CoreMatchers.instanceOf(OFSwitchHandshakeHandler.WaitInitialRoleState.class));

    OFMessage err = getBadActionErrorMessage(OFBadActionCode.BAD_TYPE, xid);

    verifyExceptionCaptured(err, SwitchStateException.class);
}

@Override
public void processMessage(Dpid dpid, OFMessage m) {
    processPacket(dpid, m);

    for (OpenFlowMessageListener listener : ofMessageListener) {
        listener.handleIncomingMessage(dpid, m);
    }
}

/**
 * We have an OFMessage we didn't expect given the current state and
 * we want to ignore the message
 * @param h the channel handler that wrote the message
 * @param m the message
 */
protected void unhandledMessageWritten(OFMessage m) {
    switchManagerCounters.unhandledMessage.increment();
    if (log.isDebugEnabled()) {
        String msg = getSwitchStateMessage(m,
                "Ignoring unexpected written message");
        log.debug(msg);
    }
}

/**
 * Get a string specifying the switch connection, state, and
 * message received. To be used as message for SwitchStateException
 * or log messages
 * @param h The channel handler (to get switch information_
 * @param m The OFMessage that has just been received
 * @param details A string giving more details about the exact nature
 * of the problem.
 * @return
 */
// needs to be protected because enum members are acutally subclasses
protected String getSwitchStateMessage(OFMessage m,
        String details) {
    return String.format("Switch: [%s], State: [%s], received: [%s]"
            + ", details: %s",
            getConnectionInfoString(),
            this.toString(),
            m.getType().toString(),
            details);
}

/**
 * Send featuresRequest msg to the switch using the handshake transactions ids.
 * @throws IOException
 */
private void sendHandshakeFeaturesRequestMessage() throws IOException {
    OFFactory factory = (ofVersion == OFVersion.OF_13) ? factory13 : factory10;
    OFMessage m = factory.buildFeaturesRequest()
            .setXid(this.handshakeTransactionIds--)
            .build();
    channel.write(Collections.singletonList(m));
}

@Override
public Command receive(IOFSwitch sw, OFMessage msg,
        FloodlightContext cntx) {
    switch (msg.getType()) {
    case PACKET_IN:
        cntIncoming.increment();
        return this.processPacketInMessage(sw, (OFPacketIn) msg, cntx);
    default:
        break;
    }
    return Command.CONTINUE;
}

private Capture<List<OFMessage>> prepareChannelForWriteList() {
    EasyMock.expect(channel.isConnected()).andReturn(Boolean.TRUE).anyTimes();
    Capture<List<OFMessage>> cMsgList = new Capture<>();
    expect(channel.write(capture(cMsgList))).andReturn(null).once();
    replay(channel);
    return cMsgList;
}

@Override
public void flush() {
    List<OFMessage> msglist = localMsgBuffer.get();
    if ((msglist != null) && (msglist.size() > 0)) {
        this.write(msglist);
        localMsgBuffer.set(null);
    }
}