RpcStreamConnection extends RpcConnection { private int continueReadIfGetPartialRead(@Nonnull final byte[] preambleBytes, final int bytesRead, @Nonnull final AtomicLong streamRecvs) throws IOException, ConnectionException { int totalBytesRead = bytesRead; while ((totalBytesRead >= 0) && (totalBytesRead < preambleBytes.length)) { int moreBytesRead = topInputStream.read(preambleBytes, totalBytesRead, preambleBytes.length - totalBytesRead); throwConnectionExceptionIfConditionFails(moreBytesRead >= 0, "server connection unexpectedly closed"); streamRecvs.incrementAndGet(); totalBytesRead += moreBytesRead; } return totalBytesRead; } RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, boolean secure);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, Socket socket);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, RpcSocketPool pool);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, Socket socket, boolean secure);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, RpcSocketPool pool, boolean secure);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, Socket socket, RpcSocketPool pool, boolean secure);  RpcStreamConnection(String serverHost, int serverPort, Properties props,
            ServerStats stats, Charset charset, 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 continueReadIfGetPartialRead() throws InvocationTargetException, IllegalAccessException, IOException { Method continueReadIfGetPartialRead = getPrivateMethod(RpcStreamConnection.class, "continueReadIfGetPartialRead", byte[].class, int.class, AtomicLong.class); byte[] preambleBytes = { 'a', 'b', 0, 0, 0 }; AtomicLong atomicLong = mock(AtomicLong.class); when(topInputStream.read(any(), anyInt(), anyInt())).thenReturn(3); int bytesRead = (int) continueReadIfGetPartialRead.invoke(mockConnection, preambleBytes, 2, atomicLong); assertThat(bytesRead, is(5)); verify(topInputStream).read(any(), anyInt(), anyInt()); }