getmessageprocessor.java

jxta-cms-src-2.4.zip 是官网上的比较稳定的CMS源代码。目前是最稳定的高版本。

/*
 * Copyright (c) 2001 Sun Microsystems, Inc.  All rights
 * reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by the
 *       Sun Microsystems, Inc. for Project JXTA."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must
 *    not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact Project JXTA at http://www.jxta.org.
 *
 * 5. Products derived from this software may not be called "JXTA",
 *    nor may "JXTA" appear in their name, without prior written
 *    permission of Sun.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
 * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 *====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of Project JXTA.  For more
 * information on Project JXTA, please see
 * <http://www.jxta.org/>.
 *
 * This license is based on the BSD license adopted by the Apache Foundation.
 *
 * $Id: GetMessageProcessor.java,v 1.14 2006/02/07 20:43:49 bondolo Exp $
 *
 */

package net.jxta.share;

import java.io.InputStream;
import java.io.IOException;

import net.jxta.endpoint.ByteArrayMessageElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.Message;

import org.apache.log4j.Logger;
import org.apache.log4j.Level;

/**
 * This class implements Get Message Processor.
 * It handles all Get requests from clients for this service.
 */

public class GetMessageProcessor extends MessageProcessor {

    private final static transient Logger LOG = Logger.getLogger(GetMessageProcessor.class.getName());

    private static int defaultChunkSize = 64 * 1024;

    public final int chunkSize;

    public GetMessageProcessor(CMS cms) {
        super(cms);
        chunkSize = defaultChunkSize;
    }

    protected void process(Message message) {
        String rid = null;
        String cid = null;

        // get the endpoint messenger for the result Peer
        Messenger messenger = null;
        try {
            String addrStr = CMS.popString(message, CMS.RETURN_ADDRESS);

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("GetMessageProcessor " + addrStr);
            }
            EndpointAddress endpointAddress = new EndpointAddress(addrStr);
            messenger = getEndpointService().getMessenger(endpointAddress);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error("could not get RETURN_ADDRESS");
            }
            return;
        }

        try {
            rid = CMS.popString(message, CMS.REQUEST_ID);
            cid = CMS.popString(message, CMS.CONTENT_ID);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.ERROR))
                LOG.error("could not get REQUEST_ID or CONTENT_ID");
            return;
        }

        long rangeBegin = 0;
        long rangeEnd = -1;
        try {
            rangeBegin = popLong(message, CMS.RANGE_BEGIN);
        } catch (IOException e) {}
        catch (NumberFormatException e) {}
        try {
            rangeEnd = popLong(message, CMS.RANGE_END);
        } catch (IOException e) {}
        catch (NumberFormatException e) {}
        if (LOG.isEnabledFor(Level.DEBUG))
            LOG.debug("RANGE_BEGIN=" + rangeBegin + ", RANGE_END=" + rangeEnd);

        // find the content
        Content content = getContent(cid);
        if (null == content) {
            return;
        }

        // get the InputStream for the content
        InputStream contentStream = null;
        try {
            contentStream = content.getInputStream();
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.ERROR))
                LOG.error("could not get content stream");
            return;
        }

        long clength = content.getContentAdvertisement().getLength();
        String clengthStr = Long.toString(clength);
        String chunkSizeStr = null;

        // zero-sized content (special case)
        if (clength == 0) {
            Message resultMsg = new Message();

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.MESSAGE_TYPE,
                    CMS.encodeAs,
                    CMS.GET_RESULT.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.REQUEST_ID,
                    CMS.encodeAs,
                    rid.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CONTENT_ID,
                    CMS.encodeAs,
                    cid.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CONTENT_LENGTH,
                    CMS.encodeAs,
                    clengthStr.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_OFFSET,
                    CMS.encodeAs,
                    "0".getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_SIZE,
                    CMS.encodeAs,
                    "0".getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_DATA,
                    CMS.encodeAs,
                    new byte[0],
                    null));

            //   resultMsg.setBytes(CMS.MESSAGE_TYPE, CMS.GET_RESULT.getBytes());
            //   resultMsg.setBytes(CMS.REQUEST_ID, rid.getBytes());
            //   resultMsg.setBytes(CMS.CONTENT_ID, cid.getBytes());
            //   resultMsg.setBytes(CMS.CONTENT_LENGTH, clengthStr.getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_OFFSET, "0".getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_SIZE, "0".getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_DATA, new byte[0]);

            // send the result
            try {
                messenger.sendMessage(resultMsg);
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.ERROR))
                    LOG.error("could not send a message");
            }

            // close the messenger
            if (null != messenger) {
                messenger = null;
            }
            return;
        }

        // send a chunk of the file at a time
        long offset = rangeBegin;
        long skip = -1;
        try {
            skip = contentStream.skip(rangeBegin);
        } catch (IOException e) {}
        if (skip != rangeBegin) {
            if (LOG.isEnabledFor(Level.DEBUG))
                LOG.debug("could not skip to rangeBegin");
            return;
        }
        if (rangeEnd < 0 || rangeEnd >= clength) {
            rangeEnd = clength - 1;
        }
        int bytesRead = 0;
        byte[] buffer;
        while (offset <= rangeEnd) {
            // create a new result message
            Message resultMsg = new Message();

            bytesRead = 0;
            try {
                int avail = contentStream.available();
                if (avail < chunkSize) {
                    buffer = new byte[avail];
                } else {
                    buffer = new byte[chunkSize];
                }

                bytesRead = contentStream.read(buffer);
                chunkSizeStr = Integer.toString(bytesRead);
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.ERROR))
                    LOG.error("could not read content stream");
                break;
            }

            try {
                Thread.sleep(15);
            } catch (InterruptedException ie) {
                LOG.warn("download interrupted");
            }

            String offsetStr = Long.toString(offset);
            offset += bytesRead;

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.MESSAGE_TYPE,
                    CMS.encodeAs,
                    CMS.GET_RESULT.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.REQUEST_ID,
                    CMS.encodeAs,
                    rid.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CONTENT_ID,
                    CMS.encodeAs,
                    cid.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CONTENT_LENGTH,
                    CMS.encodeAs,
                    clengthStr.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_OFFSET,
                    CMS.encodeAs,
                    offsetStr.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_SIZE,
                    CMS.encodeAs,
                    chunkSizeStr.getBytes(),
                    null));

            resultMsg.addMessageElement(
                new ByteArrayMessageElement(
                    CMS.CHUNK_DATA,
                    CMS.encodeAs,
                    buffer,
                    null));

            //   resultMsg.setBytes(CMS.MESSAGE_TYPE, CMS.GET_RESULT.getBytes());
            //   resultMsg.setBytes(CMS.REQUEST_ID, rid.getBytes());
            //   resultMsg.setBytes(CMS.CONTENT_ID, cid.getBytes());
            //   resultMsg.setBytes(CMS.CONTENT_LENGTH, clengthStr.getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_OFFSET, offsetStr.getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_SIZE, chunkSizeStr.getBytes());
            //   resultMsg.setBytes(CMS.CHUNK_DATA, buffer);

            // send the result
            try {
                messenger.sendMessage(resultMsg);
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.ERROR))
                    LOG.error("could not get REQUEST_ID or CONTENT_ID");
            }
        }

        // close the messenger
        if (null != messenger) {
            messenger = null;
        }
    }

    private Content getContent(String cids) {
        ContentId cid;

        try {
            cid = new ContentIdImpl(cids);
        } catch (IllegalArgumentException e) {
            if (LOG.isEnabledFor(Level.ERROR))
                LOG.error("could not create ContentId");
            return null;
        }

        Content[] clist = getContentManager().getContent(cid);
        if (clist != null && clist.length > 0) {
            return clist[0];
        }

        return null;
    }

    private static long popLong(Message message, String tag)
    throws IOException, NumberFormatException {
        String str = CMS.popString(message, tag);
        return Long.parseLong(str);
    }

}