tcpmessenger.java

来自「JXTA&#8482 is a set of open, generalize」· Java 代码 · 共 1,127 行 · 第 1/3 页

    /**
     * parses a welcome from a buffer
     *
     * @param buffer the buffer to parse, if successful the state is set to HEADER
     * @return true if successfully parsed
     */
    private boolean processWelcome(ByteBuffer buffer) {
        try {
            if (itsWelcome == null) {
                itsWelcome = new WelcomeMessage();
            }

            if (!itsWelcome.read(buffer)) {
                return false;
            }

            // The correct value for dstAddr: that of the other party.
            dstAddress = itsWelcome.getPublicAddress();
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Creating a logical address from : " + itsWelcome.getWelcomeString());
            }

            fullDstAddress = dstAddress;
            logicalDestAddress = new EndpointAddress("jxta", itsWelcome.getPeerID().getUniqueValue().toString(), null, null);

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Hello from " + itsWelcome.getPublicAddress() + " [" + itsWelcome.getPeerID() + "] ");
            }

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Registering Messenger from " + socketChannel.socket().getInetAddress().getHostAddress() + ":"+ socketChannel.socket().getPort());
            }

            try {
                // announce this messenger by connection address
                tcpTransport.messengerReadyEvent(this, getConnectionAddress());
            } catch (Throwable all) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
                }
                IOException failure = new IOException("Failure announcing messenger.");

                failure.initCause(all);
                throw failure;
            }
        } catch (IOException e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Error while parsing the welcome message", e);
            }
            closeImpl();
            return false;
        }
        return true;
    }

    /**
     * parses a header from a buffer
     *
     * @param buffer the buffer to parse, if successful the state is set to BODY
     * @return true if successfully parsed
     */
    private boolean processHeader(ByteBuffer buffer) {

        if (null == header) {
            header = new MessagePackageHeader();
        }

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine(MessageFormat.format("{0} Processing message package header, buffer stats:{1}", Thread.currentThread(), buffer.toString()));
        }
        try {
            if (!header.readHeader(buffer)) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine(MessageFormat.format("{0} maintaining current state at header, buffer stats :{1}", Thread.currentThread(), buffer.toString()));
                }
                return false;
            }
        } catch (IOException e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Error while parsing the message header", e);
            }
            if (!socketChannel.isConnected()) {
                // defunct connection close the messenger
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("SocketChannel closed. Closing the messenger");
                }
                closeImpl();
            }
        }
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine(MessageFormat.format("{0} setting current state to body, Buffer stats :{1}, remaining elements {2}:",
                            Thread.currentThread(), buffer.toString(), buffer.remaining()));
        }
        return true;
    }

    private Message processMessage(ByteBuffer buffer, MessagePackageHeader header) throws IOException {
        // TODO 20020730 [e-mail] Do something with content-coding here.
        MimeMediaType msgMime = header.getContentTypeHeader();

        return WireFormatMessageFactory.fromBuffer(buffer, msgMime, null);
    }

    /**
     * {@inheritDoc}
     * <p/>
     * This is what gets run by the Executor. It reads whatever is available,
     * processes it and then goes back to the selector waiting for more IO
     */
    public void run() {
        try {
            while (read()) {
                List<Message> msgs = processBuffer();
                for (Message msg : msgs) {
                    // Use the group's threadpool to process the message
                    tcpTransport.executor.execute(new MessageProcessor(msg));
                }
            }

            // resets the interestOPS and wakeup the selector
            if (socketChannel != null) {
                tcpTransport.register(socketChannel, this);
            }

        } catch (Throwable all) {
            if (Logging.SHOW_SEVERE) {
                LOG.log(Level.SEVERE, "Uncaught Throwable", all);
            }
        }
    }

    /**
     * @return true to indicate read maybe required
     */
    private boolean read() {
        if (closed || socketChannel == null) {
            return false;
        }
        if (!socketChannel.isConnected()) {
            closeImpl();
            return false;
        }

        try {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine(MessageFormat.format("{0} State before read(): {1}, buffer stats : {2}, remaining :{3}",
                                Thread.currentThread(), state.get(), buffer.toString(), buffer.remaining()));
            }

            int read = socketChannel.read(buffer);
            if (read < 0) {
                if (!socketChannel.isConnected() || read < 0) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine(MessageFormat.format("{0} Closing due to EOF", Thread.currentThread()));
                    }
                    closeImpl();
                }
                return false;
            } else if (read == 0) {
                return false;
            }

            tcpTransport.incrementBytesReceived(read);
            // prepare the buffer for reading
            buffer.flip();

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine(MessageFormat.format("{0} SocketChannel.read() == {1} bytes. Buffer stats:{2}, remaining {3}",
                                Thread.currentThread(), read, buffer.toString(), buffer.remaining()));
            }
            return true;
        } catch (ClosedChannelException e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Channel closed while reading data", e);
            }
            closeImpl();
            return false;
        } catch (InterruptedIOException woken) {
            // Framing is maintained within this object, therefore a read maybe interrupted then resumed
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning(MessageFormat.format("tcp receive - interrupted : read() {0} {1}:{2}", woken.bytesTransferred,
                                inetAddress.getHostAddress(), port));
            }
        } catch (IOException ioe) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "IOException occured while reading data", ioe);
            }
            closeImpl();
            return false;
        } catch (Throwable e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, MessageFormat.format("tcp receive - Error on connection {0}:{1}", inetAddress.getHostAddress(), port), e);
            }
            closingDueToFailure = true;
            closeImpl();
            return false;
        }
        // if the channel has a valid read ops return true, otherwise false
        return (socketChannel.validOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ;
    }

    /**
     * processes the input byte buffer
     * @return the list of messages present in the buffer
     */
    @SuppressWarnings("fallthrough")
    public List<Message> processBuffer() {
        
        List<Message> msgs = new ArrayList<Message>();
        boolean done = false;

        while (!done) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine(MessageFormat.format("{0} processBuffer({1}). Buffer stats:{2}, elements remaining {3}",
                                Thread.currentThread(), state.getClass(), buffer.toString(), buffer.remaining()));
            }

            switch (state.get()) {
                case WELCOME:
                    // Parse Welcome message
                    boolean wseen = processWelcome(buffer);

                    if (wseen) {
                        state.set(readState.HEADER);
                    }
                    done = true;
                    break;

                case HEADER:
                    // process the message header
                    boolean hseen = processHeader(buffer);

                    if (!hseen) {
                        done = true;
                        break;
                    }

                    receiveBeginTime = TimeUtils.timeNow();
                    // todo add a check for MTU size
                    if (header.getContentLengthHeader() > buffer.capacity()) {
                        ByteBuffer src = buffer;

                        // create an array backed buffer
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.fine(MessageFormat.format("{0} Reallocating a new buffer of size {1} to replace :{2}",
                                            Thread.currentThread(), header.getContentLengthHeader(), buffer.toString()));
                        }

                        // This implementation limits the message size to the MTU which is always < 2GB
                        buffer = ByteBuffer.allocate((int) header.getContentLengthHeader());
                        buffer.put(src);
                        buffer.flip();
                    }
                    state.set(readState.BODY);
                    /* FALLSTHROUGH */

                case BODY:
                    // process the message body
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine(MessageFormat.format(" {0} Proccessing Message Body. expecting {1}, {2} elements remaining {3}",
                                        Thread.currentThread(), header.getContentLengthHeader(), buffer.toString(), buffer.remaining()));
                    }
                    if (buffer.remaining() >= (int) header.getContentLengthHeader()) {
                        Message msg;
                        try {
                            msg = processMessage(buffer, header);
                        } catch (IOException io) {
                            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                                LOG.log(Level.FINE, "Failed to parse a message from buffer. closing connection", io);
                            }
                            closeImpl();
                            done = true;
                            break;
                        }

                        if (TransportMeterBuildSettings.TRANSPORT_METERING && (transportBindingMeter != null)) {
                            transportBindingMeter.messageReceived(initiator, msg, TimeUtils.timeNow() - receiveBeginTime,
                                    header.getContentLengthHeader());
                        }

                        tcpTransport.incrementMessagesReceived();
                        setLastUsed(TimeUtils.timeNow());
                        state.set(readState.HEADER);
                        header = null;
                        
                        msgs.add(msg);
                    } else {
                        done = true;
                        break;
                    }
            }
        } // while loop
        
        // prepare the buffer for more data
        buffer.compact();
        return msgs;
    }
    
    /**
     * A small class for processing individual messages. 
     */ 
    private class MessageProcessor implements Runnable {

        private Message msg;
        MessageProcessor(Message msg) {
            this.msg = msg;
        }
        public void run() {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine(MessageFormat.format("{0} calling EndpointService.demux({1})",
                        Thread.currentThread(),
                        msg, inetAddress.getHostAddress(), port));
            }
            tcpTransport.endpoint.demux(msg);
        }
    }

    /**
     * return the current connection status.
     *
     * @return true if there is an active connection to the remote peer, otherwise false.
     */
    private boolean isConnected() {
        return !closed;
    }

    /**
     * Return the absolute time in milliseconds at which this Connection was last used.
     *
     * @return absolute time in milliseconds.
     */
    private long getLastUsed() {
        return lastUsed;
    }

    /**
     * Set the last used time for this connection in absolute milliseconds.
     *
     * @param time absolute time in milliseconds.
     */
    private void setLastUsed(long time) {
        lastUsed = time;
    }

    /**
     * Returns the metering object for this tcpTransport
     *
     * @return the metering object for this tcpTransport
     */
    TransportBindingMeter getTransportBindingMeter() {
        return transportBindingMeter;
    }

    /**
     * Returns the remote address
     *
     * @return the remote address
     */
    private EndpointAddress getConnectionAddress() {
        // Somewhat confusing but destinationAddress is the name of that thing
        // for the welcome message.
        return itsWelcome.getDestinationAddress();
    }

    /**
     * Returns Remote PeerID
     *
     * @return Remote PeerID
     */
    private ID getDestinationPeerID() {
        return itsWelcome.getPeerID();
    }
}