jxtasocket.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

        connect(group, pipeAdv);
    }

    /**
     * Create a JxtaSocket to the given JxtaSocketAddress, within the timeout
     * specified in milliseconds.
     *
     * @param address JxtaSocket address to connect to
     * @param timeout The number of milliseconds within which the socket must
     *                be successfully created. An exception will be thrown if the socket
     *                cannot be created in the allotted time. A timeout value of {@code 0}
     *                (zero) specifies an infinite timeout.
     * @throws IOException            For failures in creating the socket.
     * @throws SocketTimeoutException If the socket cannot be created before
     *                                the timeout occurs.
     */
    public JxtaSocket(SocketAddress address, int timeout) throws IOException {
        connect(address, timeout);
    }

    /**
     * Create a JxtaSocket to any peer listening on pipeAdv this attempts
     * establish a connection to specified pipe within the context of the
     * specified group within timeout specified in milliseconds.
     *
     * @param group   group context
     * @param pipeAdv PipeAdvertisement
     * @param timeout The number of milliseconds within which the socket must
     *                be successfully created. An exception will be thrown if the socket
     *                cannot be created in the allotted time. A timeout value of {@code 0}
     *                (zero) specifies an infinite timeout.
     * @throws IOException            if an io error occurs
     * @throws SocketTimeoutException If the socket cannot be created before
     *                                the timeout occurs.
     */
    public JxtaSocket(PeerGroup group, PipeAdvertisement pipeAdv, int timeout) throws IOException {
        connect(group, pipeAdv, timeout);
    }

    /**
     * Create a JxtaSocket to any peer listening on pipeAdv
     * this attempts establish a connection to specified
     * pipe within a context of <code>group</code> and within the timeout specified in milliseconds
     *
     * @param group   group context
     * @param peerid  node to connect to
     * @param pipeAdv PipeAdvertisement
     * @param timeout The number of milliseconds within which the socket must
     *                be successfully created. An exception will be thrown if the socket
     *                cannot be created in the allotted time. A timeout value of {@code 0}
     *                (zero) specifies an infinite timeout.
     * @throws IOException            For failures in creating the socket.
     * @throws SocketTimeoutException If the socket cannot be created before
     *                                the timeout occurs.
     */
    public JxtaSocket(PeerGroup group, PeerID peerid, PipeAdvertisement pipeAdv, int timeout) throws IOException {
        connect(group, peerid, pipeAdv, timeout);
    }

    /**
     * Create a JxtaSocket to the given JxtaSocketAddress, within the timeout
     * specified in milliseconds. The JxtaSocket can be reliable (stream) or
     * not (datagram). If you want to use a SocketAddress in the constructor,
     * this is the preferred method. Either that, or use JxtaSocket(), followed
     * by create(boolean) to turn on reliability, followed by
     * connect(SocketAddress, int) or connect(SocketAddress) to make the
     * connection.
     *
     * @param address  JxtaSocket address to connect to
     * @param timeout  The number of milliseconds within which the socket must
     *                 be successfully created. An exception will be thrown if the socket
     *                 cannot be created in the allotted time. A timeout value of {@code 0}
     *                 (zero) specifies an infinite timeout.
     * @param reliable {@code true} for reliable stream connection or
     *                 {@code false} for unreliable stream connection.
     * @throws IOException            For failures in creating the socket.
     * @throws SocketTimeoutException If the socket cannot be created before
     *                                the timeout occurs.
     */
    public JxtaSocket(SocketAddress address, int timeout, boolean reliable) throws IOException {
        this.isReliable = reliable;
        connect(address, timeout);
    }

    /**
     * Create a JxtaSocket to any peer listening on pipeAdv
     * this attempts establish a connection to specified
     * pipe within a context of <code>group</code> and within the timeout specified in milliseconds
     *
     * @param group    group context
     * @param peerid   node to connect to
     * @param pipeAdv  PipeAdvertisement
     * @param timeout  The number of milliseconds within which the socket must
     *                 be successfully created. An exception will be thrown if the socket
     *                 cannot be created in the allotted time. A timeout value of {@code 0}
     *                 (zero) specifies an infinite timeout.
     * @param reliable {@code true} for reliable stream connection or
     *                 {@code false} for unreliable stream connection.
     * @throws IOException            For failures in creating the socket.
     * @throws SocketTimeoutException If the socket cannot be created before
     *                                the timeout occurs.
     */
    public JxtaSocket(PeerGroup group, PeerID peerid, PipeAdvertisement pipeAdv, int timeout, boolean reliable) throws IOException {

        this.isReliable = reliable;
        connect(group, peerid, pipeAdv, timeout);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void finalize() throws Throwable {
        if (!closed) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("JxtaSocket is being finalized without being previously closed. This is likely a users bug.");
            }
        }
        close();
        super.finalize();
    }

    /**
     * Creates either a stream or a datagram socket.
     *
     * @param reliable {@code true} for reliable stream connection or {@code false} for unreliable stream connection.
     * @throws IOException if an I/O error occurs while creating the socket.
     * @deprecated Unreliable mode is being removed. Use JxtaBiDiPipe instead.
     */
    @Deprecated
    public void create(boolean reliable) throws IOException {
        if (isBound()) {
            throw new IOException("Socket already bound, it is not possible to change connection type");
        }
        this.isReliable = reliable;
    }

    /**
     * {@inheritDoc}
     * <p/>
     * Unsupported operation, an IOException will be thrown.
     *
     * @throws IOException Thrown in all cases as this operation is not supported.
     */
    @Override
    public void bind(SocketAddress address) throws IOException {
        throw new IOException("Unsupported operation, use java.net.Socket instead");
    }

    /**
     * {@inheritDoc}
     * <p/>
     * The default connect timeout of 60 seconds is used If SocketAddress is not an instance of JxtaSocketAddress, an
     * IOException will be thrown.
     */
    @Override
    public void connect(SocketAddress address) throws IOException {
        connect(address, DEFAULT_TIMEOUT);
    }

    /**
     * {@inheritDoc}
     * <p/>
     * If SocketAddress is not an instance of JxtaSocketAddress, an IOException will be thrown.
     */
    @Override
    public void connect(SocketAddress address, int timeout) throws IOException {
        if (!(address instanceof JxtaSocketAddress)) {
            throw new IOException("Subclass of SocketAddress not supported. Use JxtaSocketAddress instead.");
        }
        JxtaSocketAddress socketAddress = (JxtaSocketAddress) address;
        PeerGroup pg = PeerGroup.globalRegistry.lookupInstance(socketAddress.getPeerGroupId());

        if (pg == null) {
            throw new IOException("Can't connect socket in PeerGroup with id " + socketAddress.getPeerGroupId()
                    + ". No running instance of the group is registered.");
        }
        connect(pg.getWeakInterface(), socketAddress.getPeerId(), socketAddress.getPipeAdv(), timeout);
        pg.unref();
    }

    /**
     * Connects to a JxtaServerSocket on any peer within the default timeout of 60 seconds
     *
     * @param group   group context
     * @param pipeAdv PipeAdvertisement
     * @throws IOException if an io error occurs
     */
    public void connect(PeerGroup group, PipeAdvertisement pipeAdv) throws IOException {
        connect(group, pipeAdv, DEFAULT_TIMEOUT);
    }

    /**
     * Connects to a JxtaServerSocket on any peer within a timeout specified in milliseconds
     *
     * @param group   group context
     * @param pipeAdv PipeAdvertisement
     * @param timeout in milliseconds
     * @throws IOException if an io error occurs
     */
    public void connect(PeerGroup group, PipeAdvertisement pipeAdv, int timeout) throws IOException {
        connect(group, null, pipeAdv, timeout);
    }

    /**
     * Connects to a JxtaServerSocket on a specific peer within a timeout specified in milliseconds
     *
     * @param group   group context
     * @param peerid  peer to connect to
     * @param pipeAdv PipeAdvertisement
     * @param timeout timeout in milliseconds
     * @throws IOException if an io error occurs
     */
    public void connect(PeerGroup group, PeerID peerid, PipeAdvertisement pipeAdv, int timeout) throws IOException {
        if (PipeService.PropagateType.equals(pipeAdv.getType())) {
            throw new IOException("Propagate pipe advertisements are not supported");
        }

        if (timeout < 0) {
            throw new IllegalArgumentException("timeout may not be negative");
        }

        this.initiator = true;
        this.group = group;
        this.remotePeerID = peerid;
        this.pipeAdv = pipeAdv;
        this.localEphemeralPipeAdv = newEphemeralPipeAdv(pipeAdv);
        this.timeout = (timeout == 0) ? Long.MAX_VALUE : timeout;

        pipeSvc = group.getPipeService();
        bind();
        Message openMsg = createConnectMessage(group, localEphemeralPipeAdv, localCredential, isReliable, initiator);
        long connectTimeoutAt = System.currentTimeMillis() + timeout;

        if (connectTimeoutAt < timeout) {
            // ensure no overflow
            connectTimeoutAt = Long.MAX_VALUE;
        }

        // Create the output pipe and send this message. Need to retry the call
        // to createOutputPipe. If there is no rendezvous yet and the
        // destination is not reachable by mcast, then createOutputPipe has no
        // effect.  We repeat it with exponential delays.
        if (peerid == null) {
            pipeSvc.createOutputPipe(pipeAdv, this);
        } else {
            pipeSvc.createOutputPipe(pipeAdv, Collections.singleton(peerid), this);
        }

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.log(Level.FINE, "Beginning Output Pipe Resolution. " + this);
        }

        // Wait for the pipe resolution.
        synchronized (pipeResolveLock) {
            while (connectOutpipe == null) {
                try {
                    long waitFor = connectTimeoutAt - System.currentTimeMillis();

                    if (waitFor <= 0) {
                        // too late
                        break;
                    }
                    if (connectOutpipe == null) {
                        // in case the pipe is resolved
                        pipeResolveLock.wait(waitFor);
                    }
                } catch (InterruptedException ie) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.log(Level.FINE, "Interrupted", ie);
                    }
                    Thread.interrupted();
                    SocketException exp = new SocketException("Connect Interrupted");
                    exp.initCause(ie);
                    throw exp;
                }
            }
        }

        if (connectOutpipe == null) {
            throw new SocketTimeoutException("Connection (resolution) timeout");
        }

        try {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Sending connect message. " + this);
            }

            // send connect message
            connectOutpipe.send(openMsg);

            // wait for the connect response.
            synchronized (socketConnectLock) {
                while (!isConnected()) {
                    try {
                        long waitFor = connectTimeoutAt - System.currentTimeMillis();
                        if (waitFor <= 0) {
                            // too late
                            break;
                        }
                        socketConnectLock.wait(waitFor);
                    } catch (InterruptedException ie) {
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.log(Level.FINE, "Interrupted", ie);
                        }
                        Thread.interrupted();
                        SocketException exp = new SocketException("Connect Interrupted");
                        exp.initCause(ie);
                        throw exp;
                    }
                }
            }
        } finally {
            connectOutpipe.close();
            connectOutpipe = null;
        }

        if (!isConnected()) {
            throw new SocketTimeoutException("Connection timeout (connect)");
        }

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("New socket connection : " + this);
        }
    }

    /**
     * obtain the cred doc from the group object
     *
     * @param group the group context
     * @return The <code>Credential</code> value
     */
    protected static Credential getDefaultCredential(PeerGroup group) {
        try {
            MembershipService membership = group.getMembershipService();
            return membership.getDefaultCredential();
        } catch (Exception e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to get credential", e);
            }
        }
        return null;
    }

    /**
     * get the remote credential doc
     *
     * @return Credential StructuredDocument
     */
    public Credential getCredentialDoc() {
        try {
            return remoteCredential;
        } catch (Exception failure) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to generate credential document ", failure);
            }
            return null;
        }
    }