class MsgStreamer extends OutputStream implements ObjToByteArraySerializer, BaseMsgStreamer, ByteBufferWriter {

  public static BaseMsgStreamer create(List<?> cons, final DistributionMessage msg,
      final boolean directReply, final DMStats stats, BufferPool bufferPool) {
    final Connection firstCon = (Connection) cons.get(0);
    // split into different versions if required
    KnownVersion version;
    final int numCons = cons.size();
    if (numCons > 1) {
      Connection con;
      Object2ObjectOpenHashMap versionToConnMap = null;
      int numVersioned = 0;
      for (Object c : cons) {
        con = (Connection) c;
        version = con.getRemoteVersion();
        if (version != null
            && KnownVersion.CURRENT_ORDINAL > version.ordinal()) {
          if (versionToConnMap == null) {
            versionToConnMap = new Object2ObjectOpenHashMap();
          }
          @SuppressWarnings("unchecked")
          ArrayList<Object> vcons = (ArrayList<Object>) versionToConnMap.get(version);
          if (vcons == null) {
            vcons = new ArrayList<Object>(numCons);
            versionToConnMap.put(version, vcons);
          }
          vcons.add(con);
          numVersioned++;
        }
      }
      if (versionToConnMap == null) {
        return new MsgStreamer(cons, msg, directReply, stats, firstCon.getSendBufferSize(),
            bufferPool);
      } else {
        // if there is a versioned stream created, then split remaining
        // connections to unversioned stream
        final ArrayList<MsgStreamer> streamers =
            new ArrayList<MsgStreamer>(versionToConnMap.size() + 1);
        final int sendBufferSize = firstCon.getSendBufferSize();
        if (numCons > numVersioned) {
          // allocating list of numCons size so that as the result of
          // getSentConnections it may not need to be reallocted later
          final ArrayList<Object> currentVersionConnections = new ArrayList<Object>(numCons);
          for (Object c : cons) {
            con = (Connection) c;
            version = con.getRemoteVersion();
            if (version == null || version.ordinal() >= KnownVersion.CURRENT_ORDINAL) {
              currentVersionConnections.add(con);
            }
          }
          streamers.add(
              new MsgStreamer(currentVersionConnections, msg, directReply, stats, sendBufferSize,
                  bufferPool));
        }
        for (ObjectIterator<Object2ObjectMap.Entry> itr =
            versionToConnMap.object2ObjectEntrySet().fastIterator(); itr.hasNext();) {
          Object2ObjectMap.Entry entry = itr.next();
          Object ver = entry.getKey();
          Object l = entry.getValue();
          streamers.add(new VersionedMsgStreamer((List<?>) l, msg, directReply, stats,
              bufferPool, sendBufferSize, (KnownVersion) ver));
        }
        return new MsgStreamerList(streamers);
      }
    } else if ((version = firstCon.getRemoteVersion()) == null) {
      return new MsgStreamer(cons, msg, directReply, stats, firstCon.getSendBufferSize(),
          bufferPool);
    } else {
      // create a single VersionedMsgStreamer
      return new VersionedMsgStreamer(cons, msg, directReply, stats, bufferPool,
          firstCon.getSendBufferSize(),
          version);
    }
  }

    MsgStreamer(List<?> cons, DistributionMessage msg, boolean directReply, DMStats stats,
      int sendBufferSize, BufferPool bufferPool);

}