/*
 * Copyright (C) 2003 Peter Jonathan Klauser. All rights reserved.
 *
 * This software is published under the terms of the Jumpi Software
 * License version 1.1, a copy of which has been included with this
 * distribution in the LICENSE-jumpi.txt file.
 *
 */

package org.jumpi.examples.jms;

import org.jumpi.Destination;
import org.jumpi.Handle;
import org.jumpi.Jumpi;
import org.jumpi.Status;
import org.jumpi.examples.JumpiClientApplication;
import org.jumpi.examples.JumpiClientApplicationRunner;
import org.jumpi.impl.connector.jms10.JmsContext;

import java.util.Hashtable;


/**
 * Receive a single serialized message over JMS, non-transactionaly,  reliably,
 * once-only-delivery.
 */
public class Replyer implements JumpiClientApplication
{
    /** The runner. */
    JumpiClientApplicationRunner runner_ = null;

    /**
     * Set the runner.
     *
     * @param runner the runner.
     */
    public void setJumpiClientApplicationRunner(
        JumpiClientApplicationRunner runner)
    {
        runner_ = runner;
    }

    /**
     * Return the configuration values to use in doWork().
     *
     * @return the configuration values to use.
     */
    public Hashtable getConfiguration()
    {
        Hashtable config = new Hashtable();
        config.put("org.jumpi.configurationloader",
            "org.jumpi.impl.ConfigurationLoader");

        config.put("org.jumpi.controller", "stateless");
        config.put("org.jumpi.controller.stateless.classname",
            "org.jumpi.impl.controller.stateless.StatelessController");

        config.put("org.jumpi.controller.stateless.connector", "jms");
        config.put("org.jumpi.controller.stateless.connector.jms.classname",
            "org.jumpi.impl.connector.jms10.JmsConnector");
        config.put("org.jumpi.controller.stateless.connector.jms.protocol",
            "jms");
        config.put("org.jumpi.controller.stateless.connector.jms.queuejndikey1",
            "java.naming.factory.initial");
        config.put("org.jumpi.controller.stateless.connector.jms.queuejndivalue1",
            "net.walend.somnifugi.SomniQueueContextFactory");
        config.put("org.jumpi.controller.stateless.connector.jms.queuejndikey2",
            "java.naming.provider.url");
        config.put("org.jumpi.controller.stateless.connector.jms.queuejndivalue2",
            "<not-used>");
        config.put("org.jumpi.controller.stateless.connector.jms.topicjndikey1",
            "java.naming.factory.initial");
        config.put("org.jumpi.controller.stateless.connector.jms.topicjndivalue1",
            "net.walend.somnifugi.SomniTopicContextFactory");
        config.put("org.jumpi.controller.stateless.connector.jms.topicjndikey2",
            "java.naming.provider.url");
        config.put("org.jumpi.controller.stateless.connector.jms.topicjndivalue2",
            "<not-used>");
        config.put("org.jumpi.controller.stateless.connector.jms.jndiqueueconnectionfactorylookup",
            "ConnectionFactory");
        config.put("org.jumpi.controller.stateless.connector.jms.jnditopicconnectionfactorylookup",
            "ConnectionFactory");

        return config;
    }

    /**
     * Run the example code.
     *
     * @param j the started, configured Jumpi instance to use in the example.
     *
     * @throws Exception if any failure condition occurs.
     */
    public void doWork(Jumpi j) throws Exception
    {
        Hashtable recvProperties = new Hashtable();
        recvProperties.put(JmsContext.PROPERTY_JMS_RECEIVE_TIMEOUT, "15000");

        Hashtable sendProperties = new Hashtable();

        Destination dest =
            j.getDestination("jms://queue:RequestorReplyerExample");

        long startTime = System.currentTimeMillis();
        long now = startTime;

        // loop for one minute
        while ((startTime + 65000) > now) {
            Handle recvHdl = j.recv(dest, recvProperties);
            runner_.evaluateHandle(recvHdl);
            recvHdl.assertStatusCode(Status.STATUS_SUCCESS,
                "Receive request failed.");

            Destination replyTo = recvHdl.getSender();
            String correlationID =
                (String) recvHdl.getProperties().get(JmsContext.PROPERTY_JMS_CORRELATION_ID);

            if (correlationID != null) {
                sendProperties.put(JmsContext.PROPERTY_JMS_CORRELATION_ID,
                    correlationID);
            } else {
                // since this would potentially be set from the loop before.
                sendProperties.remove(JmsContext.PROPERTY_JMS_CORRELATION_ID);
            }

            Handle sendHdl =
                j.send(replyTo, "Reply: " + recvHdl.getMessage(), sendProperties);
            runner_.evaluateHandle(sendHdl);
            sendHdl.assertStatusCode(Status.STATUS_SUCCESS, "Send reply failed.");

            // send 2msg/s
            Thread.currentThread().sleep(500);

            // increment the message sequence number
            now = System.currentTimeMillis();
        }
    }
}

/*
 * Version History
 * $Log: Replyer.java,v $
 * Revision 1.3  2003/06/24 22:23:13  klp
 * cosmetic javadoc correction
 *
 * Revision 1.2  2003/06/09 21:36:47  klp
 * CONFIG_LOADER_KEY refactored to be 'org.jumpi.configurationloader' property
 *
 * Revision 1.1  2003/05/20 19:46:11  klp
 * transfer to sf.net
 *
 */