jxtaserverpipe.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

                bound = false;
            }
            closed = true;
        }
    }

    /**
     *  Sets the bound attribute of the JxtaServerPipe
     */
    void setBound() {
        bound = true;
    }

    /**
     *  Gets the Timeout attribute of the JxtaServerPipe
     *
     * @return                  The soTimeout value
     * @exception  IOException  if an I/O error occurs
     */

    public synchronized int getPipeTimeout()
        throws IOException {
        if (isClosed()) {
            throw new SocketException("Server Pipe is closed");
        }

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

    /**
     *  Sets the Timeout attribute of the JxtaServerPipe
     *  a timeout of 0 blocks forever, by default this JxtaServerPipe's
     *  timeout is set to 60 seconds
     *
     * @param  timeout              The new soTimeout value
     * @exception  IOException  if an I/O error occurs
     */
    public synchronized void setPipeTimeout(int timeout)
        throws SocketException {
        if (isClosed()) {
            throw new SocketException("Server Pipe is closed");
        }

        if (timeout < 0) {
            throw new IllegalArgumentException("Negative timeout values are not allowed.");
        }
        
        if (0 == timeout) {
            this.timeout = Long.MAX_VALUE;
        } else {
            this.timeout = (long) timeout;
        }
    }

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

    }

    /**
     * Returns the binding state of the JxtaServerPipe.
     *
     * @return    true if the ServerSocket successfully bound to an address
     */
    public boolean isBound() {
        return bound;
    }

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

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

        if (message == null) {
            return;
        }

        boolean pushed;

        try {
            pushed = queue.offer(message, timeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("backlog queue full, connect request dropped", e);
            }
            
            return;
        }
        
        if (!pushed && LOG.isEnabledFor(Level.WARN)) {
            LOG.warn("backlog queue full, connect request dropped");
        }
    }
    
    /**
     * Method processMessage is the heart of this class.
     * <p>
     * This takes new incoming connect messages and constructs the JxtaBiDiPipe
     * to talk to the new client.
     * <p>
     * The ResponseMessage is created and sent.
     * 
     * @param msg The client connection request (assumed not null)
     * @return JxtaBiDiPipe Which may be null if an error occurs.
     */

    private JxtaBiDiPipe processMessage(Message msg) {

        PipeAdvertisement outputPipeAdv = null;
        PeerAdvertisement  peerAdv = null;
        StructuredDocument credDoc = null;

        try {
            MessageElement el = msg.getMessageElement(nameSpace, credTag);

            if (el != null) {
                InputStream in = el.getStream();

                credDoc = (StructuredDocument)
                        StructuredDocumentFactory.newStructuredDocument(el.getMimeType(), in);
            }

            el = msg.getMessageElement(nameSpace, reqPipeTag);

            if (el != null) {
                InputStream in = el.getStream();

                outputPipeAdv = (PipeAdvertisement)
                        AdvertisementFactory.newAdvertisement(el.getMimeType(), in);
            }

            el = msg.getMessageElement(nameSpace, remPeerTag);

            if (el != null) {
                InputStream in = el.getStream();

                peerAdv = (PeerAdvertisement)
                        AdvertisementFactory.newAdvertisement(el.getMimeType(), in);
            }

            el = msg.getMessageElement(nameSpace, reliableTag);
            boolean isReliable = false;

            if (el != null) {
                isReliable = (Boolean.valueOf((el.toString()))).booleanValue();

                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Connection request [isReliable] :" + isReliable);
                }

            }
            Messenger msgr = JxtaBiDiPipe.lightweightOutputPipe(group, outputPipeAdv, peerAdv);

            if (msgr != null) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Reliability set to :" + isReliable);
                }

                PipeAdvertisement newpipe = newInputPipe(group, outputPipeAdv);
                JxtaBiDiPipe pipe = new JxtaBiDiPipe(group, msgr, newpipe, credDoc, isReliable);

                pipe.setRemotePeerAdvertisement(peerAdv);
                pipe.setRemotePipeAdvertisement(outputPipeAdv);
                sendResponseMessage(group, msgr, newpipe);
                return pipe;
            }

        } catch (IOException e) {
            // deal with the error
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("IOException occured", e);
            }

        }
        return null;
    }

    /**
     * Method sendResponseMessage get the createResponseMessage and sends it.
     * 
     * @param group
     * @param msgr
     * @param pipeAd
     *  @throws IOException for failures sending the response message.
     */
    protected void sendResponseMessage(PeerGroup group, Messenger msgr, PipeAdvertisement pipeAd) throws IOException {

        Message msg = new Message();
        PeerAdvertisement peerAdv = group.getPeerAdvertisement();

        if (myCredentialDoc == null) {
            myCredentialDoc = JxtaBiDiPipe.getCredDoc(group);
        }

        if (myCredentialDoc != null) {
            msg.addMessageElement(JxtaServerPipe.nameSpace,
                    new InputStreamMessageElement(credTag, MimeMediaType.XMLUTF8, myCredentialDoc.getStream(), null));
        }

        msg.addMessageElement(JxtaServerPipe.nameSpace,
                new TextDocumentMessageElement(remPipeTag, (XMLDocument) pipeAd.getDocument(MimeMediaType.XMLUTF8), null));

        msg.addMessageElement(nameSpace, new TextDocumentMessageElement(remPeerTag, (XMLDocument) peerAdv.getDocument(MimeMediaType.XMLUTF8), null));
        msgr.sendMessage(msg);
    }

    /**
     * Utility method newInputPipe is used to get new pipe advertisement (w/random pipe ID) from old one.
     * <p>
     * Called by JxtaSocket to make pipe (name -> name.remote) for open message
     * <p>
     * Called by JxtaServerSocket to make pipe (name.remote -> name.remote.remote) for response message
     * 
     * @param group
     * @param pipeadv to get the basename and type from
     * @return PipeAdvertisement a new pipe advertisement
     **/
    protected static PipeAdvertisement newInputPipe(PeerGroup group, PipeAdvertisement pipeadv) {
        PipeAdvertisement adv = null;

        adv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(PipeAdvertisement.getAdvertisementType());
        adv.setPipeID(IDFactory.newPipeID((PeerGroupID) group.getPeerGroupID()));
        adv.setName(pipeadv.getName());
        adv.setType(pipeadv.getType());
        
        return adv;
    }
    
    /**
     *  get the credential doc
     *  @return Credential StructuredDocument 
     **/
    public StructuredDocument getCredentialDoc() {
        return myCredentialDoc;
    }

    /**
     *  Sets the connection credential doc
     *  If no credentials are set, the default group credential will be used
     *
     *  @param doc Credential StructuredDocument 
     **/
    public void setCredentialDoc(StructuredDocument doc) {
        this.myCredentialDoc = doc;
    }

    /**
     *  {@inheritDoc}
     *
     *  <p/>Closes the JxtaServerPipe.
     **/
    protected synchronized void finalize() throws Throwable {

        if (!closed) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("JxtaServerPipe is being finalized without being previously closed. This is likely a users bug.");
            }

        }
        
        close();
        
        super.finalize();
    }
}