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

private void processOFMultipartReply(OFStatsReply stats) {
    log.debug("Received message {} during switch-driver " +
               "subhandshake " + "from switch {} ... " +
               stats,
               getStringId());

     if (stats.getStatsType() == OFStatsType.EXPERIMENTER) {
         try {
           OFExpPortDescReply expPortDescReply =  (OFExpPortDescReply) stats;
           expPortDes.addAll(expPortDescReply.getEntries());
           if (!expPortDescReply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
               driverHandshakeComplete.set(true);
               return;
           }
          } catch (ClassCastException e) {
              log.error("Unexspected Experimenter Multipart message type {} ",
                      stats.getClass().getName());
        }
    }
}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there
 * are no more remaining. Then, pass the list to the switch for
 * parsing and configuration.
 *
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 *
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        }
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

@Override
void processOFStatisticsReply(OFChannelHandler h, OFStatsReply m)
        throws SwitchStateException {
    // Read port description
    if (m.getStatsType() != OFStatsType.PORT_DESC) {
        log.warn("Expecting port description stats but received stats "
                + "type {} from {}. Ignoring ...", m.getStatsType(),
                h.channel.getRemoteAddress());
        return;
    }
    if (m.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        log.warn("Stats reply indicates more stats from sw {} for "
                + "port description - not currently handled",
                h.getSwitchInfoString());
    }
    h.portDescReply = (OFPortDescStatsReply) m; // temp store
    log.info("Received port desc reply for switch at {}",
            h.getSwitchInfoString());
    try {
        h.sendHandshakeSetConfig();
    } catch (IOException e) {
        log.error("Unable to send setConfig after PortDescReply. "
                + "Error: {}", e.getMessage());
    }
    h.setState(WAIT_CONFIG_REPLY);
}

@Override
/**
 * Accumulate a list of the OFTableFeaturesStatsReply's until there 
 * are no more remaining. Then, pass the list to the switch for 
 * parsing and configuration.
 * 
 * The assumption is that the OFMessage dispatcher will call this each
 * time, which it does. We don't loop and receive here.
 * 
 * @param m, The potential OFTableFeaturesStatsReply message we want to include
 */
void processOFStatsReply(OFStatsReply m) {
    if (m.getStatsType() == OFStatsType.TABLE_FEATURES) {
        replies.add((OFTableFeaturesStatsReply) m);
        if (!((OFTableFeaturesStatsReply)m).getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
            handleTableFeaturesMessage(replies, false);
            nextState();
        } 
    } else {
        /* should only receive TABLE_FEATURES here */
        log.error("Received {} message but expected TABLE_FEATURES.", m.getStatsType().toString());
    }

}

private OFStatsReply getStatisticsReply(int transactionId,
        int count, boolean moreReplies, OFVersion version) {
    OFFactory factory = OFFactories.getFactory(version);

    List<OFFlowStatsEntry> statistics = new ArrayList<OFFlowStatsEntry>();
    for (int i = 0; i < count; ++i) {
        statistics.add(factory.buildFlowStatsEntry().build());
    }
    assertEquals(statistics.size(), count);

    org.projectfloodlight.openflow.protocol.OFStatsReply.Builder
            statsReplyBuilder = factory.buildFlowStatsReply()
                .setXid(transactionId)
                .setEntries(statistics);

    if (moreReplies) {
        statsReplyBuilder.setFlags(
                Collections.singleton(OFStatsReplyFlags.REPLY_MORE));
    }

    return statsReplyBuilder.build();
}

private void processOFMultipartReply(OFStatsReply stats) {
    log.debug("Received message {} during switch-driver " +
               "subhandshake " + "from switch {} ... " +
               stats,
               getStringId());

     if (stats.getStatsType() == OFStatsType.EXPERIMENTER) {
         try {
           OFExpPortDescReply expPortDescReply =  (OFExpPortDescReply) stats;
           expPortDes.addAll(expPortDescReply.getEntries());
           if (!expPortDescReply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
               driverHandshakeComplete.set(true);
               return;
           }
          } catch (ClassCastException e) {
              log.error("Unexspected Experimenter Multipart message type {} ",
                      stats.getClass().getName());
        }
    }
}

private synchronized Collection<OFFlowStatsEntry> publishFlowStats(Dpid dpid,
                                                                   OFFlowStatsReply reply) {
    //TODO: Get rid of synchronized
    fullFlowStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullFlowStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFTableStatsEntry> publishTableStats(Dpid dpid,
                                                                     OFTableStatsReply reply) {
    //TODO: Get rid of synchronized
    fullTableStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullTableStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFGroupStatsEntry> publishGroupStats(Dpid dpid,
                                                                     OFGroupStatsReply reply) {
    //TODO: Get rid of synchronized
    fullGroupStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullGroupStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFGroupDescStatsEntry> publishGroupDescStats(Dpid dpid,
                                                                             OFGroupDescStatsReply reply) {
    //TODO: Get rid of synchronized
    fullGroupDescStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullGroupDescStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFPortStatsEntry> publishPortStats(Dpid dpid,
                                                                   OFPortStatsReply reply) {
    fullPortStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullPortStats.removeAll(dpid);
    }
    return null;
}

@Override
void processOFStatisticsReply(OFChannelHandler h, OFStatsReply m)
        throws SwitchStateException {
    // Read port description
    if (m.getStatsType() != OFStatsType.PORT_DESC) {
        log.warn("Expecting port description stats but received stats "
                + "type {} from {}. Ignoring ...", m.getStatsType(),
                h.channel.getRemoteAddress());
        return;
    }
    if (m.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        log.debug("Stats reply indicates more stats from sw {} for "
                + "port description",
                h.getSwitchInfoString());
        h.portDescReplies.add((OFPortDescStatsReply)m);
        return;
    }
    else {
        h.portDescReplies.add((OFPortDescStatsReply)m);
    }
    //h.portDescReply = (OFPortDescStatsReply) m; // temp store
    log.info("Received port desc reply for switch at {}",
            h.getSwitchInfoString());
    try {
        h.sendHandshakeSetConfig();
    } catch (IOException e) {
        log.error("Unable to send setConfig after PortDescReply. "
                + "Error: {}", e.getMessage());
    }
    h.setState(WAIT_CONFIG_REPLY);
}

@Override
public void handleMessage(Dpid dpid, OFMessage msg) {
    if (isDisabled) {
        return;
    }

    try {
        switch (msg.getType()) {
            case STATS_REPLY:
                if (((OFStatsReply) msg).getStatsType() == OFStatsType.PORT) {
                    OFPortStatsReply portStatsReply = (OFPortStatsReply) msg;
                    List<OFPortStatsEntry> portStatsReplyList = portStatsReplies.get(dpid);
                    if (portStatsReplyList == null) {
                        portStatsReplyList = Lists.newCopyOnWriteArrayList();
                    }
                    portStatsReplyList.addAll(portStatsReply.getEntries());
                    portStatsReplies.put(dpid, portStatsReplyList);
                    if (!portStatsReply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
                        List<OFPortStatsEntry> statsEntries = portStatsReplies.get(dpid);
                        if (statsEntries != null) {
                            pushPortMetrics(dpid, statsEntries);
                            statsEntries.clear();
                        }
                    }
                }
                break;
            case ERROR:
                if (((OFErrorMsg) msg).getErrType() == OFErrorType.PORT_MOD_FAILED) {
                    LOG.error("port mod failed");
                }
            default:
                break;
        }
    } catch (IllegalStateException e) {
        // system is shutting down and the providerService is no longer
        // valid. Messages cannot be processed.
    }
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

private synchronized Collection<OFFlowStatsEntry> publishFlowStats(Dpid dpid,
                                                                   OFFlowStatsReply reply) {
    //TODO: Get rid of synchronized
    fullFlowStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullFlowStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFGroupStatsEntry> publishGroupStats(Dpid dpid,
                                                                  OFGroupStatsReply reply) {
    //TODO: Get rid of synchronized
    fullGroupStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullGroupStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFGroupDescStatsEntry> publishGroupDescStats(Dpid dpid,
                                                              OFGroupDescStatsReply reply) {
    //TODO: Get rid of synchronized
    fullGroupDescStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullGroupDescStats.removeAll(dpid);
    }
    return null;
}

private synchronized Collection<OFPortStatsEntry> publishPortStats(Dpid dpid,
                                                             OFPortStatsReply reply) {
    fullPortStats.putAll(dpid, reply.getEntries());
    if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
        return fullPortStats.removeAll(dpid);
    }
    return null;
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

/** write a stats request message that triggers two responses */
@Test(timeout = 5000)
public void testWriteStatsRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFFlowStatsRequest flowStatsRequest = factory.buildFlowStatsRequest().build();
    ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

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

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

    OFFlowStatsReply statsReply1 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .setFlags(Sets.immutableEnumSet(OFStatsReplyFlags.REPLY_MORE))
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply1),
            equalTo(true));
    assertThat("Future should not be complete ", future.isDone(), equalTo(false));

    OFFlowStatsReply statsReply2 = factory.buildFlowStatsReply()
            .setXid(flowStatsRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(statsReply2),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));

    assertThat(future.get(), Matchers.contains(statsReply1, statsReply2));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

public static Set<OFStatsReplyFlags> readFrom(ChannelBuffer bb) throws OFParseError {
    try {
        return ofWireValue(bb.readShort());
    } catch (IllegalArgumentException e) {
        throw new OFParseError(e);
    }
}

public static Set<OFStatsReplyFlags> ofWireValue(short val) {
    EnumSet<OFStatsReplyFlags> set = EnumSet.noneOf(OFStatsReplyFlags.class);

    if((val & REPLY_MORE_VAL) != 0)
        set.add(OFStatsReplyFlags.REPLY_MORE);
    return Collections.unmodifiableSet(set);
}

public static short toWireValue(Set<OFStatsReplyFlags> set) {
    short wireValue = 0;

    for(OFStatsReplyFlags e: set) {
        switch(e) {
            case REPLY_MORE:
                wireValue |= REPLY_MORE_VAL;
                break;
            default:
                throw new IllegalArgumentException("Illegal enum value for type OFStatsReplyFlags in version 1.2: " + e);
        }
    }
    return wireValue;
}

public static Set<OFStatsReplyFlags> readFrom(ChannelBuffer bb) throws OFParseError {
    try {
        return ofWireValue(bb.readShort());
    } catch (IllegalArgumentException e) {
        throw new OFParseError(e);
    }
}

public static Set<OFStatsReplyFlags> ofWireValue(short val) {
    EnumSet<OFStatsReplyFlags> set = EnumSet.noneOf(OFStatsReplyFlags.class);

    if((val & REPLY_MORE_VAL) != 0)
        set.add(OFStatsReplyFlags.REPLY_MORE);
    return Collections.unmodifiableSet(set);
}

public static short toWireValue(Set<OFStatsReplyFlags> set) {
    short wireValue = 0;

    for(OFStatsReplyFlags e: set) {
        switch(e) {
            case REPLY_MORE:
                wireValue |= REPLY_MORE_VAL;
                break;
            default:
                throw new IllegalArgumentException("Illegal enum value for type OFStatsReplyFlags in version 1.3: " + e);
        }
    }
    return wireValue;
}