jxtaserversocket.java

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

     * Binds the <code>JxtaServerSocket</code> to a specific pipe advertisement
     *
     * @param group   JXTA PeerGroup
     * @param pipeadv PipeAdvertisement on which pipe requests are accepted
     * @param backlog the maximum length of the queue.
     * @throws IOException if an I/O error occurs
     */
    public void bind(PeerGroup group, PipeAdvertisement pipeadv, int backlog) throws IOException {
        if (PipeService.PropagateType.equals(pipeadv.getType())) {
            throw new IOException("Propagate pipe advertisements are not supported");
        }

        if (backlog <= 0) {
            throw new IllegalArgumentException("backlog must be > 0");
        }

        this.backlog = backlog;
        queue = new ArrayBlockingQueue<Message>(backlog);
        this.group = group;
        this.pipeAdv = pipeadv;
        PipeService pipeSvc = group.getPipeService();

        serverPipe = pipeSvc.createInputPipe(pipeadv, this);
        setBound(true);
    }

    /**
     * {@inheritDoc}
     * <p/>
     * Used to bind a  JxtaServerSocket created with the no-arg constructor.
     */
    @Override
    public void bind(SocketAddress endpoint) throws IOException {
        bind(endpoint, backlog);
    }

    /**
     * {@inheritDoc}
     * <p/>
     * Used to bind a  JxtaServerSocket created with the no-arg constructor.
     */
    @Override
    public void bind(SocketAddress endpoint, int backlog) throws IOException {
        if (endpoint instanceof JxtaSocketAddress) {
            JxtaSocketAddress socketAddress = (JxtaSocketAddress) endpoint;
            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.");
            }
            bind(pg.getWeakInterface(), socketAddress.getPipeAdv(), backlog);
            pg.unref();
        } else {
            throw new IllegalArgumentException("Unsupported subclass of SocketAddress; " + "use JxtaSocketAddress instead.");
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void close() throws IOException {

        if (closed) {
            return;
        }
        closed = true;

        if (isBound()) {
            // close all the pipe
            serverPipe.close();
            setBound(false);
        }

        queue.clear();
        while (true) {
            try {
                queue.put(QUEUE_END_MESSAGE);
                // end queue message is now on the queue, we are done.
                break;
            } catch (InterruptedException woken) {
                // We MUST put the terminal message onto the queue before
                // finishing. We won't have a second chance.
                Thread.interrupted();
            }
        }
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Closed : " + this);
        }
    }

    /**
     * @return the server socket's JxtaSocketAddress
     * @see java.net.ServerSocket#getLocalSocketAddress()
     */
    @Override
    public SocketAddress getLocalSocketAddress() {
        return new JxtaSocketAddress(getGroup(), getPipeAdv());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getSoTimeout() throws IOException {
        if (isClosed()) {
            throw new SocketException("Socket is closed");
        }

        if (timeout > Integer.MAX_VALUE) {
            return 0;
        } else {
            return (int) timeout;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setSoTimeout(int timeout) throws SocketException {
        if (isClosed()) {
            throw new SocketException("Socket is closed");
        }

        if (timeout < 0) {
            throw new IllegalArgumentException("timeout must be >= 0");
        }

        if (0 == timeout) {
            this.timeout = Long.MAX_VALUE;
        } else {
            this.timeout = (long) timeout;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isBound() {
        return bound;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isClosed() {
        return closed;
    }

    /**
     * Sets whether this socket is currently bound or not. A socket is
     * considered bound if the local resources required in order to interact
     * with a remote peer are allocated and open.
     *
     * @param boundState The new bound state.
     */
    private synchronized void setBound(boolean boundState) {
        this.bound = boundState;
    }

    /**
     * Gets the group associated with this JxtaServerSocket object
     *
     * @return The group value
     */
    public PeerGroup getGroup() {
        return group;
    }

    /**
     * Gets the PipeAdvertisement associated with this JxtaServerSocket object
     *
     * @return The pipeAdv value
     */
    public PipeAdvertisement getPipeAdv() {
        return pipeAdv;
    }

    /**
     * {@inheritDoc}
     */
    public void pipeMsgEvent(PipeMsgEvent event) {

        // deal with messages as they come in
        Message message = event.getMessage();

        if (message == null) {
            return;
        }

        boolean pushed = false;
        try {
            pushed = queue.offer(message, timeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException woken) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Interrupted", woken);
            }
        }

        if (!pushed && Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
            LOG.warning("backlog queue full, connect request dropped");
        }
    }

    /**
     * processMessage is the main mechanism in establishing bi-directional connections
     * <p/>
     * It accepts connection messages and constructs a JxtaSocket with a ephemeral
     * InputPipe and a messenger.
     *
     * @param msg The client connection request (assumed not null)
     * @return JxtaSocket Which may be null if an error occurs.
     */
    private JxtaSocket processMessage(Message msg) {

        PipeAdvertisement remoteEphemeralPipeAdv = null;
        PeerAdvertisement remotePeerAdv = null;
        Credential credential = null;

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Processing a connection message : " + msg);
        }

        try {
            MessageElement el = msg.getMessageElement(MSG_ELEMENT_NAMESPACE, reqPipeTag);
            if (el != null) {
                XMLDocument pipeAdvDoc = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(el);
                remoteEphemeralPipeAdv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(pipeAdvDoc);
            }

            el = msg.getMessageElement(MSG_ELEMENT_NAMESPACE, remPeerTag);
            if (el != null) {
                XMLDocument peerAdvDoc = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(el);
                remotePeerAdv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(peerAdvDoc);
            }

            el = msg.getMessageElement(MSG_ELEMENT_NAMESPACE, credTag);
            if (el != null) {
                try {
                    XMLDocument credDoc = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(el);
                    credential = group.getMembershipService().makeCredential(credDoc);
                    if (!checkCred(credential)) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "Invalid credential");
                        }
                        return null;
                    }
                } catch (Exception ignored) {
                    // ignored
                }
            }

            boolean isReliable = false;

            el = msg.getMessageElement(MSG_ELEMENT_NAMESPACE, streamTag);
            if (el != null) {
                isReliable = Boolean.valueOf(el.toString());
            }

            if ((null != remoteEphemeralPipeAdv) && (null != remotePeerAdv)) {
                return createEphemeralSocket(group, pipeAdv, remoteEphemeralPipeAdv, remotePeerAdv, localCredential, credential, isReliable);
            } else {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.warning("Connection message did not contain valid connection information.");
                }
                return null;
            }
        } catch (IOException e) {
            // deal with the error
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "IOException occured", e);
            }
        } catch (RuntimeException e) {
            // deal with the error
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception occured", e);
            }
        }
        return null;
    }

    /**
     * Invokes the specified CredentialValidator to very a credential
     * @param cred the credential
     * @return <code>true</code> if valid, or if no validator is specified
     */
    private boolean checkCred(Credential cred) {
        return credValidator == null || credValidator.checkCred(cred);
    }

    /**
     * Construct the emphemeral socket result from accept. This method exists
     * primarily so that sub-classes can substitute a different JxtaSocket
     * sub-class.
     *
     * @param group               The peer group for the socket.
     * @param pipeAdv             The public pipe advertisement.
     * @param remoteEphemeralPipeAdv The pipe advertisement of the remote peer's
     *                            ephemeral pipe.
     * @param remotePeerAdv          The peer advertisement of the remote peer.
     * @param localCredential        Our credential.
     * @param credential          The credential of the remote peer.
     * @param isReliable          if true, uses the reliability library in non-direct mode
     * @return The new JxtaSocket instance.
     * @throws IOException if an io error occurs
     */
    protected JxtaSocket createEphemeralSocket(PeerGroup group, PipeAdvertisement pipeAdv, PipeAdvertisement remoteEphemeralPipeAdv, PeerAdvertisement remotePeerAdv, Credential localCredential, Credential credential, boolean isReliable) throws IOException {
        return new JxtaSocket(group, pipeAdv, remoteEphemeralPipeAdv, remotePeerAdv, localCredential, credential, isReliable);
    }

    /**
     * Sets the credential to be used by this socket connection. If no
     * credentials are set, the default group credential will be used.
     *
     * @param localCredential The credential to be used for connection responses
     *                     or <tt>null</tt> if the default credential is to be used.
     */
    public void setCredential(Credential localCredential) {
        this.localCredential = localCredential;
    }

    /**
     * {@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('[');
        result.append(pipeAdv.getPipeID());
        result.append(']');

        result.append(isClosed() ? " CLOSED :" : " OPEN :");
        result.append(isBound() ? " BOUND " : " UNBOUND ");
        return result.toString();
    }
}