Java example source code file (TCPChannel.java)

This example Java source code file (TCPChannel.java) is included in the alvinalexander.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Learn more about this Java project at its project page.

Java - Java tags/keywords

accesscontrolcontext, connectioexception, connection, connectionacceptor, dataoutputstream, getintegeraction, jrmp, net, reference, remoteexception, rmi, runnable, security, tcpconnection, tcpendpoint, tcptransport, threading, threads, util, weakhashmap

The TCPChannel.java Java example source code

/*
 * Copyright (c) 1996, 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
package sun.rmi.transport.tcp;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.ref.Reference;
import java.lang.ref.SoftReference;
import java.net.Socket;
import java.rmi.ConnectIOException;
import java.rmi.RemoteException;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.WeakHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import sun.rmi.runtime.Log;
import sun.rmi.runtime.NewThreadAction;
import sun.rmi.runtime.RuntimeUtil;
import sun.rmi.transport.Channel;
import sun.rmi.transport.Connection;
import sun.rmi.transport.Endpoint;
import sun.rmi.transport.TransportConstants;
import sun.security.action.GetIntegerAction;
import sun.security.action.GetLongAction;

/**
 * TCPChannel is the socket-based implementation of the RMI Channel
 * abstraction.
 *
 * @author Ann Wollrath
 */
public class TCPChannel implements Channel {
    /** endpoint for this channel */
    private final TCPEndpoint ep;
    /** transport for this channel */
    private final TCPTransport tr;
    /** list of cached connections */
    private final List<TCPConnection> freeList =
        new ArrayList<>();
    /** frees cached connections that have expired (guarded by freeList) */
    private Future<?> reaper = null;

    /** using multiplexer (for bi-directional applet communication */
    private boolean usingMultiplexer = false;
    /** connection multiplexer, if used */
    private ConnectionMultiplexer multiplexer = null;
    /** connection acceptor (should be in TCPTransport) */
    private ConnectionAcceptor acceptor;

    /** most recently authorized AccessControlContext */
    private AccessControlContext okContext;

    /** cache of authorized AccessControlContexts */
    private WeakHashMap<AccessControlContext,
                        Reference<AccessControlContext>> authcache;

    /** the SecurityManager which authorized okContext and authcache */
    private SecurityManager cacheSecurityManager = null;

    /** client-side connection idle usage timeout */
    private static final long idleTimeout =             // default 15 seconds
        AccessController.doPrivileged(
            new GetLongAction("sun.rmi.transport.connectionTimeout", 15000));

    /** client-side connection handshake read timeout */
    private static final int handshakeTimeout =         // default 1 minute
        AccessController.doPrivileged(
            new GetIntegerAction("sun.rmi.transport.tcp.handshakeTimeout",
                                 60000));

    /** client-side connection response read timeout (after handshake) */
    private static final int responseTimeout =          // default infinity
        AccessController.doPrivileged(
            new GetIntegerAction("sun.rmi.transport.tcp.responseTimeout", 0));

    /** thread pool for scheduling delayed tasks */
    private static final ScheduledExecutorService scheduler =
        AccessController.doPrivileged(
            new RuntimeUtil.GetInstanceAction()).getScheduler();

    /**
     * Create channel for endpoint.
     */
    TCPChannel(TCPTransport tr, TCPEndpoint ep) {
        this.tr = tr;
        this.ep = ep;
    }

    /**
     * Return the endpoint for this channel.
     */
    public Endpoint getEndpoint() {
        return ep;
    }

    /**
     * Checks if the current caller has sufficient privilege to make
     * a connection to the remote endpoint.
     * @exception SecurityException if caller is not allowed to use this
     * Channel.
     */
    private void checkConnectPermission() throws SecurityException {
        SecurityManager security = System.getSecurityManager();
        if (security == null)
            return;

        if (security != cacheSecurityManager) {
            // The security manager changed: flush the cache
            okContext = null;
            authcache = new WeakHashMap<AccessControlContext,
                                        Reference<AccessControlContext>>();
            cacheSecurityManager = security;
        }

        AccessControlContext ctx = AccessController.getContext();

        // If ctx is the same context as last time, or if it
        // appears in the cache, bypass the checkConnect.
        if (okContext == null ||
            !(okContext.equals(ctx) || authcache.containsKey(ctx)))
        {
            security.checkConnect(ep.getHost(), ep.getPort());
            authcache.put(ctx, new SoftReference<AccessControlContext>(ctx));
            // A WeakHashMap is transformed into a SoftHashSet by making
            // each value softly refer to its own key (Peter's idea).
        }
        okContext = ctx;
    }

    /**
     * Supplies a connection to the endpoint of the address space
     * for which this is a channel.  The returned connection may
     * be one retrieved from a cache of idle connections.
     */
    public Connection newConnection() throws RemoteException {
        TCPConnection conn;

        // loop until we find a free live connection (in which case
        // we return) or until we run out of freelist (in which case
        // the loop exits)
        do {
            conn = null;
            // try to get a free connection
            synchronized (freeList) {
                int elementPos = freeList.size()-1;

                if (elementPos >= 0) {
                    // If there is a security manager, make sure
                    // the caller is allowed to connect to the
                    // requested endpoint.
                    checkConnectPermission();
                    conn = freeList.get(elementPos);
                    freeList.remove(elementPos);
                }
            }

            // at this point, conn is null iff the freelist is empty,
            // and nonnull if a free connection of uncertain vitality
            // has been found.

            if (conn != null) {
                // check to see if the connection has closed since last use
                if (!conn.isDead()) {
                    TCPTransport.tcpLog.log(Log.BRIEF, "reuse connection");
                    return conn;
                }

                // conn is dead, and cannot be reused (reuse => false)
                this.free(conn, false);
            }
        } while (conn != null);

        // none free, so create a new connection
        return (createConnection());
    }

    /**
     * Create a new connection to the remote endpoint of this channel.
     * The returned connection is new.  The caller must already have
     * passed a security checkConnect or equivalent.
     */
    private Connection createConnection() throws RemoteException {
        Connection conn;

        TCPTransport.tcpLog.log(Log.BRIEF, "create connection");

        if (!usingMultiplexer) {
            Socket sock = ep.newSocket();
            conn = new TCPConnection(this, sock);

            try {
                DataOutputStream out =
                    new DataOutputStream(conn.getOutputStream());
                writeTransportHeader(out);

                // choose protocol (single op if not reusable socket)
                if (!conn.isReusable()) {
                    out.writeByte(TransportConstants.SingleOpProtocol);
                } else {
                    out.writeByte(TransportConstants.StreamProtocol);
                    out.flush();

                    /*
                     * Set socket read timeout to configured value for JRMP
                     * connection handshake; this also serves to guard against
                     * non-JRMP servers that do not respond (see 4322806).
                     */
                    int originalSoTimeout = 0;
                    try {
                        originalSoTimeout = sock.getSoTimeout();
                        sock.setSoTimeout(handshakeTimeout);
                    } catch (Exception e) {
                        // if we fail to set this, ignore and proceed anyway
                    }

                    DataInputStream in =
                        new DataInputStream(conn.getInputStream());
                    byte ack = in.readByte();
                    if (ack != TransportConstants.ProtocolAck) {
                        throw new ConnectIOException(
                            ack == TransportConstants.ProtocolNack ?
                            "JRMP StreamProtocol not supported by server" :
                            "non-JRMP server at remote endpoint");
                    }

                    String suggestedHost = in.readUTF();
                    int    suggestedPort = in.readInt();
                    if (TCPTransport.tcpLog.isLoggable(Log.VERBOSE)) {
                        TCPTransport.tcpLog.log(Log.VERBOSE,
                            "server suggested " + suggestedHost + ":" +
                            suggestedPort);
                    }

                    // set local host name, if unknown
                    TCPEndpoint.setLocalHost(suggestedHost);
                    // do NOT set the default port, because we don't
                    // know if we can't listen YET...

                    // write out default endpoint to match protocol
                    // (but it serves no purpose)
                    TCPEndpoint localEp =
                        TCPEndpoint.getLocalEndpoint(0, null, null);
                    out.writeUTF(localEp.getHost());
                    out.writeInt(localEp.getPort());
                    if (TCPTransport.tcpLog.isLoggable(Log.VERBOSE)) {
                        TCPTransport.tcpLog.log(Log.VERBOSE, "using " +
                            localEp.getHost() + ":" + localEp.getPort());
                    }

                    /*
                     * After JRMP handshake, set socket read timeout to value
                     * configured for the rest of the lifetime of the
                     * connection.  NOTE: this timeout, if configured to a
                     * finite duration, places an upper bound on the time
                     * that a remote method call is permitted to execute.
                     */
                    try {
                        /*
                         * If socket factory had set a non-zero timeout on its
                         * own, then restore it instead of using the property-
                         * configured value.
                         */
                        sock.setSoTimeout((originalSoTimeout != 0 ?
                                           originalSoTimeout :
                                           responseTimeout));
                    } catch (Exception e) {
                        // if we fail to set this, ignore and proceed anyway
                    }

                    out.flush();
                }
            } catch (IOException e) {
                if (e instanceof RemoteException)
                    throw (RemoteException) e;
                else
                    throw new ConnectIOException(
                        "error during JRMP connection establishment", e);
            }
        } else {
            try {
                conn = multiplexer.openConnection();
            } catch (IOException e) {
                synchronized (this) {
                    usingMultiplexer = false;
                    multiplexer = null;
                }
                throw new ConnectIOException(
                    "error opening virtual connection " +
                    "over multiplexed connection", e);
            }
        }
        return conn;
    }

    /**
     * Free the connection generated by this channel.
     * @param conn The connection
     * @param reuse If true, the connection is in a state in which it
     *        can be reused for another method call.
     */
    public void free(Connection conn, boolean reuse) {
        if (conn == null) return;

        if (reuse && conn.isReusable()) {
            long lastuse = System.currentTimeMillis();
            TCPConnection tcpConnection = (TCPConnection) conn;

            TCPTransport.tcpLog.log(Log.BRIEF, "reuse connection");

            /*
             * Cache connection; if reaper task for expired
             * connections isn't scheduled, then schedule it.
             */
            synchronized (freeList) {
                freeList.add(tcpConnection);
                if (reaper == null) {
                    TCPTransport.tcpLog.log(Log.BRIEF, "create reaper");

                    reaper = scheduler.scheduleWithFixedDelay(
                        new Runnable() {
                            public void run() {
                                TCPTransport.tcpLog.log(Log.VERBOSE,
                                                        "wake up");
                                freeCachedConnections();
                            }
                        }, idleTimeout, idleTimeout, TimeUnit.MILLISECONDS);
                }
            }

            tcpConnection.setLastUseTime(lastuse);
            tcpConnection.setExpiration(lastuse + idleTimeout);
        } else {
            TCPTransport.tcpLog.log(Log.BRIEF, "close connection");

            try {
                conn.close();
            } catch (IOException ignored) {
            }
        }
    }

    /**
     * Send transport header over stream.
     */
    private void writeTransportHeader(DataOutputStream out)
        throws RemoteException
    {
        try {
            // write out transport header
            DataOutputStream dataOut =
                new DataOutputStream(out);
            dataOut.writeInt(TransportConstants.Magic);
            dataOut.writeShort(TransportConstants.Version);
        } catch (IOException e) {
            throw new ConnectIOException(
                "error writing JRMP transport header", e);
        }
    }

    /**
     * Use given connection multiplexer object to obtain new connections
     * through this channel.
     */
    synchronized void useMultiplexer(ConnectionMultiplexer newMultiplexer) {
        // for now, always just use the last one given
        multiplexer = newMultiplexer;

        usingMultiplexer = true;
    }

    /**
     * Accept a connection provided over a multiplexed channel.
     */
    void acceptMultiplexConnection(Connection conn) {
        if (acceptor == null) {
            acceptor = new ConnectionAcceptor(tr);
            acceptor.startNewAcceptor();
        }
        acceptor.accept(conn);
    }

    /**
     * Closes all the connections in the cache, whether timed out or not.
     */
    public void shedCache() {
        // Build a list of connections, to avoid holding the freeList
        // lock during (potentially long-running) close() calls.
        Connection[] conn;
        synchronized (freeList) {
            conn = freeList.toArray(new Connection[freeList.size()]);
            freeList.clear();
        }

        // Close all the connections that were free
        for (int i = conn.length; --i >= 0; ) {
            Connection c = conn[i];
            conn[i] = null; // help gc
            try {
                c.close();
            } catch (java.io.IOException e) {
                // eat exception
            }
        }
    }

    private void freeCachedConnections() {
        /*
         * Remove each connection whose time out has expired.
         */
        synchronized (freeList) {
            int size = freeList.size();

            if (size > 0) {
                long time = System.currentTimeMillis();
                ListIterator<TCPConnection> iter = freeList.listIterator(size);

                while (iter.hasPrevious()) {
                    TCPConnection conn = iter.previous();
                    if (conn.expired(time)) {
                        TCPTransport.tcpLog.log(Log.VERBOSE,
                            "connection timeout expired");

                        try {
                            conn.close();
                        } catch (java.io.IOException e) {
                            // eat exception
                        }
                        iter.remove();
                    }
                }
            }

            if (freeList.isEmpty()) {
                reaper.cancel(false);
                reaper = null;
            }
        }
    }
}

/**
 * ConnectionAcceptor manages accepting new connections and giving them
 * to TCPTransport's message handler on new threads.
 *
 * Since this object only needs to know which transport to give new
 * connections to, it doesn't need to be per-channel as currently
 * implemented.
 */
class ConnectionAcceptor implements Runnable {

    /** transport that will handle message on accepted connections */
    private TCPTransport transport;

    /** queue of connections to be accepted */
    private List<Connection> queue = new ArrayList<>();

    /** thread ID counter */
    private static int threadNum = 0;

    /**
     * Create a new ConnectionAcceptor that will give connections
     * to the specified transport on a new thread.
     */
    public ConnectionAcceptor(TCPTransport transport) {
        this.transport = transport;
    }

    /**
     * Start a new thread to accept connections.
     */
    public void startNewAcceptor() {
        Thread t = AccessController.doPrivileged(
            new NewThreadAction(ConnectionAcceptor.this,
                                "Multiplex Accept-" + ++ threadNum,
                                true));
        t.start();
    }

    /**
     * Add connection to queue of connections to be accepted.
     */
    public void accept(Connection conn) {
        synchronized (queue) {
            queue.add(conn);
            queue.notify();
        }
    }

    /**
     * Give transport next accepted connection, when available.
     */
    public void run() {
        Connection conn;

        synchronized (queue) {
            while (queue.size() == 0) {
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                }
            }
            startNewAcceptor();
            conn = queue.remove(0);
        }

        transport.handleMessages(conn, true);
    }
}