public static BaseMsgStreamer create(List<Connection> cons, final DistributionMessage msg,
      final boolean directReply, final DMStats stats, BufferPool bufferPool) {
    final Connection firstCon = cons.get(0);

    // split into different versions if required
    final int numCons = cons.size();
    if (numCons > 1) {
      Object2ObjectOpenHashMap<KnownVersion, List<Connection>> versionToConnMap = null;
      int numVersioned = 0;
      for (final Connection connection : cons) {
        final KnownVersion version = connection.getRemoteVersion();
        if (version != null
            && KnownVersion.CURRENT_ORDINAL > version.ordinal()) {
          if (versionToConnMap == null) {
            versionToConnMap = new Object2ObjectOpenHashMap<>();
          }
          versionToConnMap.computeIfAbsent(version, k -> new ArrayList<>(numCons)).add(connection);
          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 List<MsgStreamer> streamers = new ArrayList<>(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 reallocated later
          final List<Connection> currentVersionConnections = new ArrayList<>(numCons);
          for (Connection connection : cons) {
            final KnownVersion version = connection.getRemoteVersion();
            if (version == null || version.ordinal() >= KnownVersion.CURRENT_ORDINAL) {
              currentVersionConnections.add(connection);
            }
          }
          streamers.add(
              new MsgStreamer(currentVersionConnections, msg, directReply, stats, sendBufferSize,
                  bufferPool));
        }
        for (ObjectIterator<Object2ObjectMap.Entry<KnownVersion, List<Connection>>> itr =
            versionToConnMap.object2ObjectEntrySet().fastIterator(); itr.hasNext();) {
          Object2ObjectMap.Entry<KnownVersion, List<Connection>> entry = itr.next();
          streamers.add(new VersionedMsgStreamer(entry.getValue(), msg, directReply, stats,
              bufferPool, sendBufferSize, entry.getKey()));
        }
        return new MsgStreamerList(streamers);
      }
    } else {
      final KnownVersion version = firstCon.getRemoteVersion();
      if (null == version) {
        return new MsgStreamer(cons, msg, directReply, stats, firstCon.getSendBufferSize(),
            bufferPool);
      } else {
        return new VersionedMsgStreamer(cons, msg, directReply, stats, bufferPool,
            firstCon.getSendBufferSize(), version);
      }
    }
  }
Create message streamers splitting into versioned streamers, if required, for given list of
connections to remote nodes. This method can either return a single MsgStreamer object or a
List of MsgStreamer objects.
geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java