jxtamulticastsocket.java

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

                LOG.warn("failed to get credential", e);
            }
        }
        return null;
    }

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

    /**
     * Closes this MutlicastSocket.
     *
     *@exception  IOException  if an I/O error occurs when closing this
     *      socket.
     */
    public synchronized void close() {
        if (closed) {
            return;
        }
        bound = false;
        closed = true;
        in.close();
        outputPipe.close();
        queue.close();
        in = null;
    }

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

        Message message = event.getMessage();
        if (message == null) {
            return;
        }

        MessageElement element = null;
        // does the message contain any data
        element = (MessageElement)
                  message.getMessageElement(NAMESPACE, DATATAG);
        if (element == null) {
            return;
        }
        try {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Pushing a message onto queue");
            }
            queue.push(message, -1);
        } catch (InterruptedException e) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Interrupted", e);
            }
        }
    }

    /**
     *  Gets the Timeout attribute of the JxtaMulticastSocket
     *
     * @return                  The soTimeout value
     * @exception  IOException  if an I/O error occurs
     */
    public synchronized int getSoTimeout() {
        return timeout;
    }

    /**
     *  Sets the Timeout attribute of the JxtaMulticastSocket
     *  a timeout of 0 blocks forever, by default this Socket's
     *  timeout is set to 0
     *
     * @param  timeout              The new soTimeout value
     * @exception  IOException  if an I/O error occurs
     */
    public synchronized void setSoTimeout(int timeout)
    throws SocketException {
        checkState();
        this.timeout = timeout;
    }

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

    /**
     * Throws a SocketException if closed or not bound
     */
    private void checkState() throws SocketException {
        if (isClosed()) {
            throw new SocketException("MulticastSocket is closed");
        } else if (!isBound()) {
            throw new SocketException("MulticastSocket not bound");
        }
    }

    /**
     * {@inheritDoc}
     */
    public void send(DatagramPacket packet) throws IOException {
        checkState();
        byte [] data = packet.getData();
        InputStream bais = new ByteArrayInputStream(data, packet.getOffset(), packet.getLength());

        Message msg = new Message();
        msg.addMessageElement(NAMESPACE,
                              srcElement);
        msg.addMessageElement(NAMESPACE,
                              new InputStreamMessageElement(DATATAG,
                                                            MimeMediaType.AOS,
                                                            bais,
                                                            null));
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Sending a data packet");
        }
        InetAddress address = packet.getAddress();
        PeerID pid = null;
        if (address != null) {
            String pidStr = address.getHostName();
            try {
                pid =(PeerID) IDFactory.fromURI(new URI(pidStr));
            } catch (Exception ex) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                     LOG.debug("Invalid source PeerID multicasting instead");
                }
            }
        }
        if (pid != null) {
            // Unicast datagram
            // create a op pipe to the destination peer
            OutputPipe op = pipeSvc.createOutputPipe(pipeAdv, Collections.singleton(pid), 1000);
            op.send(msg);
            op.close();
        } else {
            // multicast
            outputPipe.send(msg);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void receive(DatagramPacket packet) throws IOException {
        checkState();
        Message msg = null;
        //data
        MessageElement del = null;
        //src
        MessageElement sel = null;
        try {
            msg = (Message) queue.pop(timeout);
            if(msg == null) {
                if (timeout > 0) {
                    throw new SocketTimeoutException("Socket timeout reached");
                } else {
                    return;
                }
            }
            del = msg.getMessageElement(NAMESPACE, DATATAG);
            sel = msg.getMessageElement(NAMESPACE, SRCIDTAG);
            if (del == null || sel == null) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Message contains no data element, returning");
                }
                return;
            } else {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Popped a message off the queue");
                }
            }
        } catch (InterruptedException e) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Exception occured", e);
            }
            throw new IOException(e.toString());
        }
        if (del.getByteLength() > packet.getLength()) {
            throw new IOException("Datagram can not accomodate message of size :"+ del.getByteLength());
        }
        String addrStr = new String(sel.getBytes(false),
                                    0,
                                    (int) sel.getByteLength(),
                                    "UTF8");
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Src Address :"+addrStr);
        }
        InetAddress address = InetAddress.getByAddress(addrStr, fauxip);
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Setting Data, and Src Address :"+address);
        }
        packet.setAddress(address);
        packet.setData(del.getBytes(false));
    }

    /**
     * {@inheritDoc}
     */
    public InetAddress getLocalAddress() {
        if (isClosed()) {
            return null;
        }
        return localAddress;
    }

    /**
     * {@inheritDoc}
     */
    public SocketAddress getLocalSocketAddress() {
        if (isClosed()) {
            return null;
        }
        return socketAddress;
    }

    /**
     * {@inheritDoc}
     */
    public void bind(SocketAddress addr) throws SocketException {
        if (isBound()) {
            throw new SocketException("Already bound");
        }
    }
}