udpccmessenger.java

High performance DB query

            }
        } else if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(data, "Dropped Callback",
                                                  null, null));
        }
    }

    /**
     * Method serialize_size
     * Only Called from NetworkCore
     *
     * @param msg
     * @return
     */
    public int serialize_size(Object msg) {
        if (msg instanceof Payload) {
            return SerializationManager.getPayloadSize((Payload) msg);
        } else {
            return maxSize;
        }
    }

    /**
     * Method serialize
     * Only Called from NetworkCore
     *
     * @param msg
     * @param buf
     */
    public void serialize(Object msg, ByteBuffer buf) {
        Integer bufferID = StructuredLogMessage.getReference();

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(msg,
                                                  "Beginning Serialization",
                                                  new Object[]{"b",
                                                               bufferID}, null));
        }

        try {
            if (internalSerialization) {
                SerializationManager.serialize(new DataByteBuffer(buf),
                                               (Payload) msg);
                buf.limit(buf.position());
            } else {
                byteStream.reset();

                ObjectOutputStream objectStream =
                    new ObjectOutputStream(byteStream);

                objectStream.writeObject(msg);

                byte[] bytes = byteStream.toByteArray();

                // Place data in ByteBuffer
                buf.put(bytes);
                buf.limit(buf.position());
            }
        } catch (Exception exception) {
            logger.error(
                new StructuredLogMessage(
                    msg, "Serialization Error", null, null), exception);
        }

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(msg,
                                                  "Completed Serialization",
                                                  new Object[]{"z",
                                                               bufferID}, null));
        }

        StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.UDP_NETWORK,
                                StatVars.UDP_DATA, buf.limit());
    }

    /**
     * Method deserialize
     * Only Called from NetworkCore
     *
     * @param buf
     * @return
     * @throws Exception
     */
    public Object deserialize(ByteBuffer buf) throws Exception {
        Integer bufferID = StructuredLogMessage.getReference();

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(bufferID,
                                                  "Beginning Deserialization",
                                                  new Object[]{"z",
                                                               String.valueOf(buf.limit()
                                                               - buf.position())}, null));
        }

        Object msg = null;

        try {
            if (internalSerialization) {
                msg = SerializationManager.deSerialize(new DataByteBuffer(buf));
            } else {
                byte bytes[] = new byte[buf.limit() - buf.position()];

                buf.get(bytes);

                ByteArrayInputStream byteStream =
                    new ByteArrayInputStream(bytes);
                ObjectInputStream ois = new ObjectInputStream(byteStream);

                msg = ois.readObject();
            }

            if (Output.debuggingEnabled) {
                logger.debug(
                    new StructuredLogMessage(
                        bufferID, "Completed Deserialization", new Object[]{"m",
                                                                            Integer.toHexString(
                                                                            (msg
                                                                            == null)
                                                                            ? 0
                                                                            : msg.hashCode())}, null));
            }
        } catch (Exception exception) {
            logger.error(
                new StructuredLogMessage(
                    bufferID, "Deserialization Error", null, null), exception);
        }

        StatCollector.addSample(StatVars.NETWORK_IN, StatVars.UDP_NETWORK,
                                StatVars.UDP_DATA, buf.limit());

        return msg;
    }

    /**
     * Method cb
     * Only Called from NetworkCore
     *
     * @param user_data
     * @param success
     */
    public void cb(Object user_data, boolean success) {
        if (user_data != null) {
            if (Output.debuggingEnabled) {
                logger.debug(new StructuredLogMessage(user_data,
                                                      "UDPCC Status Callback",
                                                      new Object[]{"s",
                                                                   String.valueOf(
                                                                   success)}, new Object[]{
                                                                       "u",
                                                                       user_data}));
            }

            eventCore.register_timer(
                0, this,
                (new AckCallbackObject((AckObject) user_data, success)));
        }
    }

    /**
     * Method recv
     * Only Called from NetworkCore
     *
     * @param msg
     * @param src
     * @param local
     * @param tries
     * @param wait_ms
     * @param est_rtt_ms
     */
    public void recv(Object msg, InetSocketAddress src,
                     InetSocketAddress local, int tries, long wait_ms,
                     long est_rtt_ms) {
        UDPClient client =
            (UDPClient) listenersClients.get(new Integer(local.getPort()));

        if (client != null) {
            if (Output.debuggingEnabled) {
                logger.debug(
                    new StructuredLogMessage(
                        msg, "Network Message Received", new Object[]{"t",
                                                                      String.valueOf(
                                                                      tries),
                                                                      "w",
                                                                      String.valueOf(
                                                                      wait_ms),
                                                                      "r",
                                                                      String.valueOf(
                                                                      est_rtt_ms)}, null));
            }

            if (msg != null) {
                eventCore.register_timer(0, this,
                                         (new RunObject(client,
                                                        local.getPort(), src,
                                                        (Payload) msg)));
            }
        }
    }

    /**
     * Class SendObject
     *
     */
    public class SendObject {

        public InetSocketAddress source;
        public InetSocketAddress destination;
        public Payload payload;
        public AckObject ackObject;

        /**
         * Constructor SendObject
         *
         * @param source
         * @param destination
         * @param payload
         * @param ackObject
         */
        public SendObject(InetSocketAddress source,
                          InetSocketAddress destination, Payload payload,
                          AckObject ackObject) {
            this.source = source;
            this.destination = destination;
            this.payload = payload;
            this.ackObject = ackObject;
        }
    }


    /**
     * Class ReleaseObject
     *
     */
    public class ReleaseObject {

        public UdpCC listener;

        /**
         * Constructor ReleaseObject
         *
         * @param listener
         */
        public ReleaseObject(UdpCC listener) {
            this.listener = listener;
        }
    }


    /**
     * Class AckCallbackObject
     *
     */
    public class AckCallbackObject {

        public AckObject ackObject;
        public boolean success;

        /**
         * Constructor AckCallbackObject
         *
         * @param ackObject
         * @param success
         */
        public AckCallbackObject(AckObject ackObject, boolean success) {
            this.ackObject = ackObject;
            this.success = success;
        }
    }


    /**
     * Class RunObject
     *
     */
    public class RunObject {

        public UDPClient client;
        public int localPort;
        public InetSocketAddress remoteIPAddress;
        public Payload data;

        /**
         * Constructor RunObject
         *
         * @param client
         * @param localPort
         * @param remoteIPAddress
         * @param data
         */
        public RunObject(UDPClient client, int localPort,
                         InetSocketAddress remoteIPAddress, Payload data) {
            this.client = client;
            this.localPort = localPort;
            this.remoteIPAddress = remoteIPAddress;
            this.data = data;
        }
    }


    /**
     * Method processNetworkCoreSend
     *
     * @param sendObject
     */
    public void processNetworkCoreSend(SendObject sendObject) {
        if (sendObject.source == null) {
            sender.send(sendObject.payload, sendObject.destination, timeout,
                        this, sendObject.ackObject);

            if (Output.debuggingEnabled) {
                logger.debug(
                    new StructuredLogMessage(
                        sendObject.payload, "Passed to UDPCC Default Sender",
                        null, null));
            }
        } else {
            UdpCC theSender = (UdpCC) sendChannels.get(sendObject.source);

            if (theSender == null) {
                theSender = new UdpCC(networkCore, sendObject.source, this,
                                      this);

                theSender.set_debug_level(debugLevel);
                sendChannels.put(sendObject.source, theSender);
            }

            theSender.send(sendObject.payload, sendObject.destination, timeout,
                           this, sendObject.ackObject);

            if (Output.debuggingEnabled) {
                logger.debug(
                    new StructuredLogMessage(
                        sendObject.payload,
                        "Passed to UDPCC Non-Default Sender", null, null));
            }
        }
    }

    /**
     * Method processNetworkCoreRelease
     *
     * @param releaseObject
     */
    public void processNetworkCoreRelease(ReleaseObject releaseObject) {
        releaseObject.listener.close();
    }

    /**
     * Method processEventCoreAckCallback
     *
     * @param ackCallbackObject
     */
    public void processEventCoreAckCallback(
            AckCallbackObject ackCallbackObject) {
        if ((ackCallbackObject.ackObject != null)
                && (ackCallbackObject.ackObject.client != null)) {
            ackCallbackObject.ackObject.client.handleUDPNetworkAck(
                ackCallbackObject.ackObject.ackData, ackCallbackObject.success);
        }
    }

    /**
     * Method processEventCoreRun
     *
     * @param runObject
     */
    public void processEventCoreRun(RunObject runObject) {
        runEvent(runObject.client, runObject.localPort,
                 runObject.remoteIPAddress, runObject.data);
    }

    /**
     * Method timer_cb
     *
     * @param user_data
     */
    public void timer_cb(Object user_data) {
        if (user_data instanceof SendObject) {
            processNetworkCoreSend((SendObject) user_data);
        }

        if (user_data instanceof ReleaseObject) {
            processNetworkCoreRelease((ReleaseObject) user_data);
        }

        if (user_data instanceof AckCallbackObject) {
            processEventCoreAckCallback((AckCallbackObject) user_data);
        }

        if (user_data instanceof RunObject) {
            processEventCoreRun((RunObject) user_data);
        }
    }
}