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); 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 { final ArrayList<MsgStreamer> streamers = new ArrayList<MsgStreamer>(versionToConnMap.size() + 1); final int sendBufferSize = firstCon.getSendBufferSize(); if (numCons > numVersioned) { 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 { return new VersionedMsgStreamer(cons, msg, directReply, stats, bufferPool, firstCon.getSendBufferSize(), version); } } MsgStreamer(List<?> cons, DistributionMessage msg, boolean directReply, DMStats stats,
      int sendBufferSize, BufferPool bufferPool); @Override ConnectExceptions getConnectExceptions(); @Override List<?> getSentConnections(); static BaseMsgStreamer create(List<?> cons, final DistributionMessage msg,
      final boolean directReply, final DMStats stats, BufferPool bufferPool); @Override void reserveConnections(long startTime, long ackTimeout, long ackSDTimeout); @Override int writeMessage(); @Override void write(int b); @Override void flush(); void realFlush(boolean lastFlushForMessage); @Override void close(); @Override void write(byte[] source, int offset, int len); @Override void write(ByteBuffer bb); @Override void writeBoolean(boolean v); @Override void writeByte(int v); @Override void writeShort(int v); @Override void writeChar(int v); @Override void writeInt(int v); @Override void writeLong(long v); @Override void writeFloat(float v); @Override void writeDouble(double v); @Override void writeBytes(String str); @Override void writeChars(String s); @Override void writeUTF(String str); @Override void writeAsSerializedByteArray(Object v);  }
@Test public void create() { final BaseMsgStreamer msgStreamer = createMsgStreamer(false); assertThat(msgStreamer).isInstanceOf(MsgStreamer.class); }