RpcStreamConnection extends RpcConnection { private int continueReadIfIncompleteRead(@Nonnull final AtomicLong streamRecvs, final int payloadLength, @Nonnull final byte[] packetBytes, final int packetBytesRead) throws IOException, ConnectionException { int totalPacketBytesRead = packetBytesRead; while (totalPacketBytesRead < payloadLength) { stats.incompleteReads.incrementAndGet(); int moreBytesRead = topInputStream.read(packetBytes, totalPacketBytesRead, payloadLength - totalPacketBytesRead); throwConnectionExceptionIfConditionFails(moreBytesRead >= 0, "Perforce server network connection closed unexpectedly"); streamRecvs.incrementAndGet(); stats.totalBytesRecv.getAndAdd(moreBytesRead); totalPacketBytesRead += moreBytesRead; } return totalPacketBytesRead; } RpcStreamConnection(String serverHost, int serverPort, Properties props,
                               ServerStats stats, P4Charset p4Charset, boolean secure); private  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, P4Charset p4Charset, Socket socket, boolean secure); private  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, P4Charset p4Charset, Socket socket, RpcSocketPool pool, boolean secure);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, P4Charset p4Charset, Socket socket, RpcSocketPool pool, boolean secure,
            String rsh); String getServerIpPort(); String getClientIpPort(); void disconnect(final RpcPacketDispatcher dispatcher); RpcPacket getRpcPacket(); RpcPacket getRpcPacket(final RpcPacketFieldRule fieldRule,
            final IFilterCallback filterCallback); int getSystemRecvBufferSize(); int getSystemSendBufferSize(); long putRpcPackets(@Nonnull RpcPacket[] packets); long putRpcPacket(@Nonnull RpcPacket packet); @Override void useConnectionCompression(); static final String TRACE_PREFIX; }
@Test public void continueReadIfIncompleteRead() throws InvocationTargetException, IllegalAccessException, IOException { Method continueReadIfIncompleteRead = getPrivateMethod(RpcStreamConnection.class, "continueReadIfIncompleteRead", AtomicLong.class, int.class, byte[].class, int.class); int packetBytesRead = 2; int payloadLength = 5; when(topInputStream.read(any(byte[].class), anyInt(), anyInt())).thenReturn(3); int bytesRead = (int) continueReadIfIncompleteRead.invoke(mockConnection, new AtomicLong(0), payloadLength, new byte[] { 'a', 'b', 0, 0, 0 }, packetBytesRead); assertThat(bytesRead, is(5)); verify(topInputStream).read(any(byte[].class), anyInt(), anyInt()); }