jxtasocket.java

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

    /**
     * {@inheritDoc}
     */
    @Override
    public void setSoTimeout(int soTimeout) throws SocketException {
        if (soTimeout < 0) {
            throw new IllegalArgumentException("Invalid Socket timeout :" + soTimeout);
        }

        this.timeout = soTimeout;
        if (!isBound()) {
            return;
        }

        // If we are bound then set the timeout on the streams.
        // FIXME, ros does not define a timeout as it only relies on window saturation, it should take into account
        // the socket timeout
        if (isReliable) {
            if (ris != null) {
                ris.setTimeout(soTimeout);
            }
        } else {
            nonReliableInputStream.setTimeout((long) soTimeout);
        }
    }

    /**
     * Gets the Maximum Retry Timeout of the reliability layer
     *
     * @return The maximum retry Timeout value
     * @deprecated The reliability layer manages it's own maximum. This value is not useful.
     */
    @Deprecated
    public int getMaxRetryTimeout() {
        return maxRetryTimeout;
    }

    /**
     * Gets the Maximum Retry Timeout of the reliability layer
     *
     * @param maxRetryTimeout The new maximum retry timeout value
     * @throws IllegalArgumentException if maxRetryTimeout exceeds jxta platform maximum retry timeout
     * @deprecated The reliability layer manages it's own maximum. This value is not useful.
     */
    @Deprecated
    public void setMaxRetryTimeout(int maxRetryTimeout) {
        if (maxRetryTimeout <= 0 || maxRetryTimeout > MAXRETRYTIMEOUT) {
            throw new IllegalArgumentException("Invalid Maximum retry timeout :" + maxRetryTimeout + " Exceed Global maximum retry timeout :"
                    + MAXRETRYTIMEOUT);
        }
        this.maxRetryTimeout = maxRetryTimeout;
    }

    /**
     * Gets the Retry Timeout of the reliability layer
     *
     * @return The retry Timeout value
     */
    public int getRetryTimeout() {
        return retryTimeout;
    }

    /**
     * Sets the Retry Timeout of the underlying reliability layer.
     * In reliable mode it is possible for this call to block
     * trying to obtain a lock on reliable input stream
     *
     * @param retryTimeout The new retry timeout value
     * @throws SocketException if an I/O error occurs
     */
    public void setRetryTimeout(int retryTimeout) throws SocketException {
        if (retryTimeout <= 0 || retryTimeout > maxRetryTimeout) {
            throw new IllegalArgumentException("Invalid Retry Socket timeout :" + retryTimeout);
        }
        this.retryTimeout = retryTimeout;
        if (outgoing != null) {
            outgoing.setTimeout(retryTimeout);
        }
    }

    /**
     * When in reliable mode, gets the Reliable library window size
     *
     * @return The windowSize value
     */
    public int getWindowSize() {
        return windowSize;
    }

    /**
     * When in reliable mode, sets the Reliable library window size
     *
     * @param windowSize The new window size value
     * @throws SocketException if an I/O error occurs
     */
    public void setWindowSize(int windowSize) throws SocketException {
        if (isBound()) {
            throw new SocketException("Socket bound. Can not change the window size");
        }
        this.windowSize = windowSize;
    }

    /**
     * Returns the closed state of the JxtaSocket.
     *
     * @return true if the socket has been closed
     */
    @Override
    public boolean isClosed() {
        return closed;
    }

    /**
     * Performs on behalf of JxtaSocketOutputStream.
     *
     * @param buf    the data.
     * @param offset the start offset in the data.
     * @param length the number of bytes to write.
     * @throws IOException if an I/O error occurs
     * @see java.io.OutputStream#write
     */
    protected void write(byte[] buf, int offset, int length) throws IOException {
        checkState();
        if (isReliable) {
            ros.write(buf, offset, length);
        } else {
            byte[] bufCopy = new byte[length];
            System.arraycopy(buf, offset, bufCopy, 0, length);

            Message msg = new Message();
            msg.addMessageElement(JxtaServerSocket.MSG_ELEMENT_NAMESPACE,
                    new ByteArrayMessageElement(JxtaServerSocket.dataTag, MimeMediaType.AOS, bufCopy, 0, length, null));
            remoteEphemeralPipeMsgr.sendMessageB(msg, null, null);
        }
    }

    /**
     * @throws SocketException if closed, not bound or not connected.
     */
    private void checkState() throws SocketException {
        if (isClosed()) {
            throw new SocketException("Socket is closed.");
        } else if (!isBound()) {
            throw new SocketException("Socket not bound.");
        } else if (!isConnected()) {
            throw new SocketException("Socket not connected.");
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getSendBufferSize() throws SocketException {
        if (isOutputShutdown()) {
            throw new SocketException("Socket is closed");
        }
        return (outputBufferSize == -1) ? DEFAULT_OUTPUT_BUFFER_SIZE : outputBufferSize;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setSendBufferSize(int size) throws SocketException {
        if (isOutputShutdown()) {
            throw new SocketException("Socket is closed");
        }

        if (size < 1) {
            throw new IllegalArgumentException("negative/zero buffer size");
        }

        if ((null != remoteEphemeralPipeMsgr) && (size > remoteEphemeralPipeMsgr.getMTU())) {
            throw new IllegalArgumentException("Buffer size larger than limit : " + remoteEphemeralPipeMsgr.getMTU());
        }

        outputBufferSize = size;
        if (null != ros) {
            try {
                ros.setSendBufferSize(size);
            } catch (SocketException failure) {
                throw failure;
            } catch (IOException failed) {
                SocketException failure = new SocketException("Failed");

                failure.initCause(failed);
                throw failure;
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getReceiveBufferSize() throws SocketException {
        if (isInputShutdown()) {
            throw new SocketException("Socket is closed");
        }
        // this is just rough size
        return outputBufferSize * windowSize;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean getKeepAlive() throws SocketException {
        if (inputShutdown) {
            throw new SocketException("Socket is closed");
        }
        return false;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getTrafficClass() throws SocketException {
        throw new SocketException("TrafficClass not yet defined");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setTrafficClass(int tc) throws SocketException {
        // a place holder when and if we decide to add hints regarding
        // flow info hints such as (IPTOS_LOWCOST (0x02), IPTOS_RELIABILITY (0x04), etc
        throw new SocketException("TrafficClass not yet defined");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isInputShutdown() {
        return inputShutdown;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isOutputShutdown() {
        return outputShutdown;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void sendUrgentData(int data) throws IOException {
        throw new SocketException("Urgent data not supported");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setOOBInline(boolean state) throws SocketException {
        throw new SocketException("Enable/disable OOBINLINE supported");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setKeepAlive(boolean state) throws SocketException {
        if (isClosed()) {
            throw new SocketException("Socket is closed");
        }
        throw new SocketException("Operation not supported");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void shutdownInput() throws IOException {
        inputShutdown = true;
        if (isReliable) {
            // hard close (EOF on next read)
            ris.close();
        } else {
            // hard close (EOF on next read)
            nonReliableInputStream.close();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void shutdownOutput() throws IOException {
        outputShutdown = true;
        if (isReliable) {
            ros.setLingerDelay(timeout);
            // soft close (finish sending if you can)
            ros.close();
        } else {
            // soft close (finish sending if you can)
            nonReliableOutputStream.close();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public SocketAddress getLocalSocketAddress() {
        if (!isBound()) {
            return null;
        }
        return new JxtaSocketAddress(group, localEphemeralPipeAdv, group.getPeerAdvertisement());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public SocketAddress getRemoteSocketAddress() {
        if (!isConnected()) {
            return null;
        }
        return new JxtaSocketAddress(group, remoteEphemeralPipeAdv, remotePeerAdv);
    }

    /**
     * {@inheritDoc}
     * <p/>
     * This output is suitable for debugging but should not be parsed. All
     * of the information is available through other means.
     */
    @Override
    public String toString() {

        StringBuilder result = new StringBuilder();

        result.append(getClass().getName());
        result.append('@');
        result.append(System.identityHashCode(this));
        result.append('[');

        if (null != pipeAdv) {
            result.append(pipeAdv.getPipeID().getUniqueValue());
        }
        result.append('/');

        if (null != localEphemeralPipeAdv) {
            result.append(localEphemeralPipeAdv.getPipeID().getUniqueValue());
        }
        result.append(']');
        result.append(isClosed() ? " CLOSED :" : " OPEN :");
        result.append(initiator ? " I " : " i ");
        result.append(