asynctcprawmessenger.java

High performance DB query

/*
 * @(#)$Id: AsyncTCPRawMessenger.java,v 1.8 2004/09/15 18:05:32 huebsch Exp $
 *
 * Copyright (c) 2001-2004 Regents of the University of California.
 * All rights reserved.
 *
 * This file is distributed under the terms in the attached BERKELEY-LICENSE
 * file. If you do not find these files, copies can be found by writing to:
 * Computer Science Division, Database Group, Universite of California,
 * 617 Soda Hall #1776, Berkeley, CA 94720-1776. Attention: Berkeley License
 *
 * Copyright (c) 2003-2004 Intel Corporation. All rights reserved.
 *
 * This file is distributed under the terms in the attached INTEL-LICENSE file.
 * If you do not find these files, copies can be found by writing to:
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300,
 * Berkeley, CA, 94704.  Attention:  Intel License Inquiry.
 */


package runtime.services.network.tcpraw;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import org.apache.log4j.Logger;
import runtime.schedulers.ASyncCore;
import services.Output;
import services.network.tcpraw.TCPRawClient;
import services.network.tcpraw.TCPRawConnection;
import services.network.tcpraw.TCPRawMessenger;
import util.logging.LogMessage;
import util.logging.StructuredLogMessage;

/**
 * Class AsyncTCPRawMessenger
 *
 */
public class AsyncTCPRawMessenger
        implements TCPRawMessenger, ASyncCore.SelectableCB {

    private static Logger logger = Logger.getLogger(AsyncTCPRawMessenger.class);
    private ASyncCore networkCore;
    private ASyncCore eventCore;
    private InetAddress localAddress;
    private HashMap listeners;

    /**
     * Constructor AsyncTCPRawMessenger
     *
     * @param theSyncObject
     * @param networkCore
     * @param eventCore
     * @param localAddress
     * @param maxBufferSize
     */
    public AsyncTCPRawMessenger(Object theSyncObject, ASyncCore networkCore,
                                ASyncCore eventCore, InetAddress localAddress,
                                int maxBufferSize) {
        this.networkCore = eventCore;
        this.eventCore = eventCore;
        this.localAddress = localAddress;
        this.listeners = new HashMap();
    }

    /**
     * Method convertStringToBuffer
     *
     * @param string
     * @return
     */
    public ByteBuffer convertStringToBuffer(String string) {
        ByteBuffer buffer = ByteBuffer.wrap(string.getBytes());

        buffer.position(0);

        return buffer;
    }

    /**
     * Method listen
     *
     * @param portNumber
     * @param client
     * @return
     */
    public boolean listen(Integer portNumber, TCPRawClient client) {
        if (listeners.get(portNumber) == null) {
            try {
                ServerSocketChannel newListener = ServerSocketChannel.open();

                newListener.configureBlocking(false);

                InetSocketAddress localSocketAddress =
                    new InetSocketAddress(localAddress, portNumber.intValue());

                newListener.socket().bind(localSocketAddress);

                SelectionKey selectionKey =
                    networkCore.register_selectable(newListener,
                                                    SelectionKey.OP_ACCEPT,
                                                    this, portNumber);

                listeners.put(portNumber,
                              new ListenerInfo(newListener, client,
                                               selectionKey));

                return true;
            } catch (Exception e) {
                logger.error(
                    new LogMessage(new Object[]{"Unable to listen on port."}),
                    e);

                return false;
            }
        } else {
            return false;
        }
    }

    /**
     * Method release
     *
     * @param portNumber
     */
    public void release(Integer portNumber) {
        ListenerInfo listenerInfo = (ListenerInfo) listeners.remove(portNumber);

        if (listenerInfo != null) {
            networkCore.unregister_selectable(listenerInfo.selectionKey);

            try {
                listenerInfo.serverSocketChannel.close();
            } catch (Exception e) {
                logger.error(
                    new LogMessage(
                        new Object[]{
                            "Unable to release server socket channel."}), e);
            }
        }
    }

    /**
     * Method connect
     *
     * @param destination
     * @param client
     * @return
     */
    public TCPRawConnection connect(InetSocketAddress destination,
                                    TCPRawClient client) {
        try {
            SocketChannel newChannel = SocketChannel.open();

            newChannel.configureBlocking(false);
            newChannel.connect(destination);

            AsyncTCPRawConnection newConnection =
                new AsyncTCPRawConnection(newChannel, networkCore, client);

            return newConnection;
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * Method disconnect
     *
     * @param connection
     */
    public void disconnect(TCPRawConnection connection) {
        connection.disconnect();
    }

    /**
     * Method select_cb
     *
     * @param key
     * @param user_data
     */
    public void select_cb(SelectionKey key, Object user_data) {
        ListenerInfo listenerInfo = (ListenerInfo) listeners.get(user_data);

        if (listenerInfo != null) {
            try {
                SocketChannel newChannel =
                    listenerInfo.serverSocketChannel.accept();

                if (newChannel != null) {
                    newChannel.configureBlocking(false);

                    AsyncTCPRawConnection newConnection =
                        new AsyncTCPRawConnection(newChannel, networkCore,
                                                  listenerInfo.client);
                    Integer eventID = StructuredLogMessage.getReference();

                    if (Output.debuggingEnabled) {
                        logger.debug(
                            new StructuredLogMessage(
                                eventID,
                                "Beginning New TCPRaw Connection Handler",
                                null, null));
                    }

                    listenerInfo.client.handleTCPRawNetworkNew(newConnection);

                    if (Output.debuggingEnabled) {
                        logger.debug(
                            new StructuredLogMessage(
                                eventID,
                                "Completed New TCPRaw Connection Handler",
                                null, null));
                    }
                }
            } catch (Exception e) {
                logger.error(
                    new LogMessage(new Object[]{"Unable to accept channel."}),
                    e);
            }
        }
    }

    private class ListenerInfo {

        public ServerSocketChannel serverSocketChannel;
        public TCPRawClient client;
        public SelectionKey selectionKey;

        /**
         * Constructor ListenerInfo
         *
         * @param serverSocketChannel
         * @param client
         * @param selectionKey
         */
        public ListenerInfo(ServerSocketChannel serverSocketChannel,
                            TCPRawClient client, SelectionKey selectionKey) {
            this.serverSocketChannel = serverSocketChannel;
            this.client = client;
            this.selectionKey = selectionKey;
        }
    }
}