Stream bytes read event handler.

@param conn
Connection
@param channel
Channel
@param source
Header
@param streamBytesRead
Bytes read event context
191948
protected void onStreamBytesRead(RTMPConnection conn, Channel channel, Header source, BytesRead streamBytesRead) {
        conn.receivedBytesRead(streamBytesRead.getBytesRead());
    }
1044622
@Override
    public VirtualConnection read(long numBytes, TCPReadCompletedCallback readCallback, boolean forceQueue, int timeout) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "read(" + numBytes + ",..," + forceQueue + "," + timeout + ")");
        }
        getTCPConnLink().incrementNumReads();
        if (getConfig().getDumpStatsInterval() > 0) {
            getTCPConnLink().getTCPChannel().totalAsyncReads.incrementAndGet();
        }
        // always check external call parms
        checkForErrors(numBytes, true, timeout);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
            Socket socket = getTCPConnLink().getSocketIOChannel().getSocket();
            Tr.event(tc, "read (async) requested for local: " + socket.getLocalSocketAddress() + " remote: " + socket.getRemoteSocketAddress());
        }

        VirtualConnection vc = null;

        if (isAborted()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Previously aborted, unable to perform read");
            }
            // async errors can only be passed through the callback, which means
            // if they didn't provide one then we have no way of reporting the
            // failure
            if (null != readCallback) {
                IOException ioe = new IOException("Connection aborted by program");
                readCallback.error(getTCPConnLink().getVirtualConnection(), this, ioe);
            }
        } else {
            vc = readInternal(numBytes, readCallback, forceQueue, timeout);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "read: " + vc);
        }
        return vc;
    }
430126
public int read(byte []buffer, int offset, int length)
    throws IOException
  {
    return _stream.read(buffer, offset, length);
  }
1044089
private boolean readMessage(ByteBuffer buffer) {

    // logger.finest("readMessage: " + bytesRead + " bytes read, remaining=" +
    // remainingBytes.length + ", state=" + state.toString() + ", recurse=" +
    // recurse);

    buffer.position(0);
    int bytesWeHaveBeenReading = 0;

    switch (state) {
      case SERVICELIST:
        StringBuilder builder = new StringBuilder();
        builder.append(buffer.asCharBuffer());
        parseServiceList(builder.toString());
        buffer.position(0);// reset position!
        bytesWeHaveBeenReading = buffer.limit(); // bytesWeHaveBeenreading = all
        // bytes in buffer!
        break;

      case HANDSHAKE:
        if (buffer.limit() >= 6) {
          byte[] dst = new byte[6];
          buffer.position(0);// read from start!
          buffer.get(dst);
          buffer.position(0);// reset position!
          bytesWeHaveBeenReading = 6; // 6 bytes will be removed from buffer
          String handShake = new String(dst);
          if (!handShake.equals("STP/1\n")) {
            close();
            connectionHandler.onException(
                new CommunicationException("Expected STP/1, got: " + handShake));
          }
          setState(State.EMPTY);
          connectionHandler.onHandshake(true);
        }
        break;

      case EMPTY: // read 4 byte header: STP/0
        if (buffer.limit() >= 4) {
          byte[] headerPrefix = new byte[4];
          buffer.get(headerPrefix);
          buffer.position(0);
          bytesWeHaveBeenReading = 4;
          ByteString incomingPrefix = ByteString.copyFrom(headerPrefix);
          if (stpPrefix.equals(incomingPrefix)) {
            setState(State.STP);
            /*
            if(buffer.hasRemaining()){
                buffer.compact();
                readMessage(buffer, buffer.position(), true);
            } else {
                buffer.clear();
            }
            */
          } else {
            close();
            connectionHandler.onException(
                new CommunicationException("Expected empty header"));
          }
        }
        break;

      case STP:
        // Try to read size
        buffer.position(0);
        if (buffer.limit() <= 0) {
          logger.finest("STP: Empty buffer");
          break;
        }
        int messageSize = readRawVarint32(buffer);// read part of buffer
        bytesWeHaveBeenReading = buffer.position();
        buffer.position(0);

        // If we got size, read more, if not just leave it!
        if (buffer.limit() >= bytesWeHaveBeenReading + messageSize) {
          buffer.position(bytesWeHaveBeenReading);

          // Read type and Payload!
          int messageType = buffer.get();
          bytesWeHaveBeenReading += 1;

          byte[] payload = new byte[--messageSize];
          buffer.get(payload);
          buffer.position(0);

          bytesWeHaveBeenReading += messageSize; // 32 bits = 4 bytes :-)

          setState(State.EMPTY);

          try {
            processMessage(messageType, payload);
          } catch (IOException e) {
            close();
            connectionHandler.onException(new CommunicationException(
                "Error while processing the message: " + e.getMessage()));
          }
        } else {
          // 4 + messageSize because of the int at the beginning
          logger.finest(String.format("Tried to read a message and expected %d bytes, but got %d",
                                      (4 + messageSize), buffer.limit()));

          buffer.position(0);
          bytesWeHaveBeenReading = 0;
        }
        break;

    }

    // Pop number of read bytes from
    if (bytesWeHaveBeenReading > 0) {

      // Pop X bytes, and keep message for the rest
      int rest = buffer.limit() - bytesWeHaveBeenReading;
      if (rest <= 0) {
        buffer.clear();
        buffer.limit(0);
      } else {
        byte[] temp = new byte[rest];

        buffer.position(bytesWeHaveBeenReading);
        buffer.get(temp, 0, rest);

        buffer.clear();
        buffer.limit(rest);
        buffer.position(0);
        buffer.put(temp, 0, rest);
        buffer.position(0);// set position back to start!
      }

      logger.finest(String.format("Did read message of %d bytes, new buffer size is %d",
                                  bytesWeHaveBeenReading, buffer.limit()));

      return true; // We did read a message :-)
    } else {
      if (buffer.limit() > 0) {
        logger.finest("Did NOT read message from buffer of size " + buffer.limit());
      } else {
        logger.finest("No messages in empty buffer");
      }
      return false;
    }
  }
937357
private void write(IRTMPEvent event, Number streamId) {
        log.trace("write to stream id: {} channel: {}", streamId, id);
        final Header header = new Header();
        final Packet packet = new Packet(header, event);
        // set the channel id
        header.setChannelId(id);
        int ts = event.getTimestamp();
        if (ts != 0) {
            header.setTimer(event.getTimestamp());
        }
        header.setStreamId(streamId);
        header.setDataType(event.getDataType());
        // should use RTMPConnection specific method.. 
        //log.trace("Connection type for write: {}", connection.getClass().getName());
        connection.write(packet);
    }
1067892
public static ByteBuffer blockingRead(SocketChannel so, long timeout, byte[] bytes) throws IOException
	{
		ByteBuffer b = ByteBuffer.wrap(bytes);

		if (bytes.length == 0)
			return b;

		final long timeoutTime = (timeout > 0) ? (System.currentTimeMillis() + timeout) : (Long.MAX_VALUE);

		while (b.remaining() != 0 && System.currentTimeMillis() < timeoutTime)
		{
			if (!so.isConnected())
				throw new IOException("Socket closed during read operation!");

			so.read(b);

			if (b.remaining() != 0)
			{
				// sleep for a short time
				try
				{
					Thread.sleep(20);
				}
				catch (InterruptedException e)
				{
				}
			}
		}

		if (System.currentTimeMillis() >= timeoutTime)
		{
			return null;
		}

		b.rewind(); // make it easy for the caller to read from the buffer (if they're interested)

		return b;
	}
1125742
private void receiveConnecting(ClientSocket client, DataInputStream buffer, byte from, StateConnection expected)
            throws IOException
    {
        if (ServerImpl.checkValidity(client, from, expected))
        {
            // Receive the name
            final byte[] name = new byte[buffer.readByte()];
            if (buffer.read(name) == -1)
            {
                throw new IOException("Unable to read client name !");
            }
            client.setName(new String(name, NetworkMessage.CHARSET));

            // Send new state
            client.setState(StateConnection.CONNECTED);
            client.getOut().writeByte(NetworkMessageSystemId.CONNECTED);
            client.getOut().writeByte(client.getId());
            client.getOut().writeByte(clientsNumber - 1);

            // Send the list of other clients
            for (final ClientSocket other : clients.values())
            {
                if (other.getId() != from)
                {
                    ServerImpl.writeIdAndName(client, other.getId(), other.getName());
                }
            }
            // Send message of the day if has
            if (messageOfTheDay != null)
            {
                final byte[] motd = messageOfTheDay.getBytes(NetworkMessage.CHARSET);
                client.getOut().writeByte(motd.length);
                client.getOut().write(motd);
            }
            // Send
            client.getOut().flush();
        }
    }
960330
@Handler
    public void onOpenConnection(OpenTcpConnection event) {
        try {
            SocketChannel socketChannel = SocketChannel.open(event.address());
            channels.add(new TcpChannelImpl(socketChannel));
        } catch (ConnectException e) {
            fire(new ConnectError(event, "Connection refused.", e));
        } catch (IOException e) {
            fire(new ConnectError(event, "Failed to open TCP connection.", e));
        }
    }
582963
private int read(byte[] bytes, int offset, int length, Duration receiveTimeout)
            throws InterruptedException, TTransportException
    {
        long timeRemaining = receiveTimeout.roundTo(TimeUnit.NANOSECONDS);
        lock.lock();
        try {
            while (!closed) {
                int bytesAvailable = readBuffer.readableBytes();
                if (bytesAvailable > 0) {
                    int begin = readBuffer.readerIndex();
                    readBuffer.readBytes(bytes, offset, Math.min(bytesAvailable, length));
                    int end = readBuffer.readerIndex();
                    return end - begin;
                }
                if (timeRemaining <= 0) {
                    break;
                }
                timeRemaining = condition.awaitNanos(timeRemaining);
                if (exception != null) {
                    try {
                        throw new TTransportException(exception);
                    }
                    finally {
                        exception = null;
                        closed = true;
                        close();
                    }
                }
            }
            if (closed) {
                throw new TTransportException("channel closed !");
            }
        }
        finally {
            lock.unlock();
        }
        throw new TTransportException(String.format("receive timeout, %d ms has elapsed", receiveTimeout.toMillis()));
    }
384183
public static void transfer(InputStream in, byte[] bytes) throws IOException
    {
        // Read in the bytes
        int offset = 0;
        int numRead;
        while (offset < bytes.length && (numRead = in.read(bytes, offset, bytes.length - offset)) >= 0)
        {
            offset += numRead;
        }

        if (offset < bytes.length)
        {
            throw new IOException("Retry:  Not all bytes were transferred correctly.");
        }
    }
63267
@SuppressWarnings("unchecked")
    public VirtualConnection read(long numBytes, WsByteBuffer[] requestBuffers) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "read entry: stream " + myID + " request: " + requestBuffers + " num bytes requested: " + numBytes);
        }

        long streamByteCount = streamReadSize; // number of bytes available on this stream
        long requestByteCount = bytesRemaining(requestBuffers); // total capacity of the caller byte array
        int reqArrayIndex = 0; // keep track of where we are in the caller's array

        // if at least numBytes are ready to be processed as part of the HTTP Request/Response, then load it up
        if (numBytes > streamReadSize || requestBuffers == null) {
            if (this.headersCompleted) {
                return h2HttpInboundLinkWrap.getVirtualConnection();
            } else {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "read exit: stream " + myID + " more bytes requested than available for read: " + numBytes + " > " + streamReadSize);
                }
                return null;
            }
        }

        if (streamByteCount < requestByteCount) {
            // the byte count requested exceeds the caller array capacity
            actualReadCount = streamByteCount;
        } else {
            actualReadCount = requestByteCount;
        }

        // copy bytes from this stream into the caller arrays
        for (int bytesRead = 0; bytesRead < actualReadCount; bytesRead++) {

            // find the first buffer from the caller that has remaining capacity
            while ((requestBuffers[reqArrayIndex].position() == requestBuffers[reqArrayIndex].limit())) {
                reqArrayIndex++;
            }

            // find the next stream buffer that has data
            while (!streamReadReady.isEmpty() && !streamReadReady.get(0).hasRemaining()) {
                streamReadReady.get(0).release();
                streamReadReady.remove(0);
            }
            requestBuffers[reqArrayIndex].put(streamReadReady.get(0).get());
        }

        // put stream array back in shape
        streamReadSize = 0;
        readLatch = new CountDownLatch(1);
        for (WsByteBuffer buffer : ((ArrayList<WsByteBuffer>) streamReadReady.clone())) {
            streamReadReady.clear();
            if (buffer.hasRemaining()) {
                moveDataIntoReadBufferArray(buffer.slice());
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "read exit: " + streamId());
        }
        // return the vc since this was a successful read
        return h2HttpInboundLinkWrap.getVirtualConnection();
    }
1139527
private void readFully(InputStream in, byte[] bytes) throws IOException {
        int read = ByteStreams.read(in, bytes, 0, bytes.length);
        if (read == 0) {
            throw new NoBytesReadException();
        } else if (read != bytes.length) {
            throw new EOFException("reached end of stream after reading "
                    + read + " bytes; " + bytes.length + " bytes expected");
        }
    }
440014
@Override
    public synchronized void processConnect(WebSocketChannel channel, WSURI uri, String[] protocols) {
        LOG.entering(CLASS_NAME, "processConnect", new Object[] { uri, protocols });

        try {
            WebSocketNativeChannel nativeChannel = (WebSocketNativeChannel)channel;
            if (nativeChannel.getProxy() != null) {
                throw new IllegalStateException("Bridge proxy previously set");
            }
            
            Proxy proxy = BridgeUtil.createProxy(uri.getURI(), this);
            proxy.setPeer(channel);
            nativeChannel.setProxy(proxy);
            
            String[] params;
            if (protocols != null) {
                String s = "";
                for (int i=0; i<protocols.length; i++) {
                    if (i>0) {
                        s += ",";
                    }
                    s += protocols[i];
                }
                params = new String[] { "WEBSOCKET", uri.toString(), s, ""};
            } else {
                params = new String[] { "WEBSOCKET", uri.toString() };
            }
            proxy.processEvent(XoaEventKind.CREATE, params);
        }
        catch (Exception e) {
            LOG.log(Level.FINE, "While initializing WebSocket proxy: "+e.getMessage(), e);
            listener.connectionFailed(channel, e);
        }
    }
782994
void process(ByteChannel sock, AutoBuffer ab) throws IOException {
        int requestType = ab.get1();
        switch (requestType) {
            case CREATE_FRAME:
                ExternalFrameWriterBackend.handleWriteToChunk(sock, ab);
                break;
            case DOWNLOAD_FRAME:
                ExternalFrameReaderBackend.handleReadingFromChunk(sock, ab);
                break;
        }
    }
53430
void sendHeaderAndCounterNameAndRead(Channel channel) {
      ByteBuf buf = getHeaderAndCounterNameBufferAndRead(channel, 0);
      channel.writeAndFlush(buf);
   }
404615
public static final long read(ReadableByteChannel channel, ByteBuffer[] dsts) throws IOException
    {
        return read(channel, dsts, 0, dsts.length);
    }
291224
@Override
    protected int readBytesWireFormat ( byte[] buffer, int bufferIndex ) throws SMBProtocolDecodingException {
        int structureSize = SMBUtil.readInt2(buffer, bufferIndex);
        if ( structureSize != 2 ) {
            throw new SMBProtocolDecodingException("Expected structureSize = 2");
        }
        return 2;
    }