/*
 *
 *   Copyright (c) 1999 - 2011 my-Channels Ltd
 *   Copyright (c) 2012 - 2017 Software AG, Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA, and/or its subsidiaries and/or its affiliates and/or their licensors.
 *
 *   Use, reproduction, transfer, publication or disclosure is prohibited except as specifically provided for in your License Agreement with Software AG.
 *
 */
package com.pcbsys.nirvana.apps;

import java.io.IOException;
import java.util.Vector;

import com.pcbsys.nirvana.client.*;

public class requester extends nSampleApp implements nEventListener, Runnable {

  private static requester myself;
  private String username;
  Object lock = new Object();
  private boolean async = false;
  private nQueue reqQueue;
  private nQueue respQueue;
  private boolean transactional = false;
  private nQueueReader queueReader;

  public static void main(String[] args) {

    myself = new requester();

    //Process input variables.
    myself.processArgs(args);
    requester.processEnvironmentVariables();
    String reqQueueName = System.getProperty("REQUESTQUEUENAME");
    String respQueueName = System.getProperty("RESPONSEQUEUENAME");
    myself.username = System.getProperty("USERNAME");
    if (System.getProperty("ASYNC") != null) {
      myself.async = System.getProperty("ASYNC").equalsIgnoreCase("true");
    }
    if (System.getProperty("TRANSACTIONAL") != null) {
      myself.transactional = System.getProperty("TRANSACTIONAL").equalsIgnoreCase("true");
    }

    //Construct realm properties.
    String[] rproperties = new String[4];
    rproperties = parseRealmProperties(System.getProperty("RNAME"));
    System.out.println("username: " + myself.username);
    myself.doit(rproperties, reqQueueName, respQueueName);
  }

  @Override
  protected void processArgs(String[] args) {
    if (args.length > 4) {
      Usage();
      UsageEnv();
    } else {
      switch (args.length) {
        case 4:
          System.getProperties().put("TRANSACTIONAL", args[3]);
        case 3:
          System.getProperties().put("ASYNC", args[2]);
        case 2:
          System.getProperties().put("RESPONSEQUEUENAME", args[1]);
        case 1:
          if (args[0].equals("-?")) {
            Usage();
            UsageEnv();
          }
          System.getProperties().put("REQUESTQUEUENAME", args[0]);
      }
    }
  }

  private void Usage() {
    System.out.println("Usage ...\n");
    System.out.println("requester <request queue> <request queue>\n");

    System.out.println("<Required Arguments> \n");
    System.out.println("<request queue> - Queue onto which request are published");
    System.out.println("<response queue> - Queue onto which responses are published");
    System.out.println("\n[Optional Arguments] \n");
    System.out
        .println("[asynchronous] - Whether to use asynchronous producing and consuming - true/false, default false.");
    System.out.println(
        "[transactional] - Whether to use transactional production and consumption of events - true/false, default false.");
    System.out.println("\n\nNote: -? provides help on environment variables \n");
  }

  public void doit(String[] rproperties, String requestQueueName, String responseQueueName) {
    try {
      myself.constructSession(rproperties);

      //Connect to response queue.
      nChannelAttributes respAtr = new nChannelAttributes();
      respAtr.setName(responseQueueName);
      respQueue = mySession.findQueue(respAtr);

      //Connect to request queue.
      nChannelAttributes reqAtr = new nChannelAttributes();
      reqAtr.setName(requestQueueName);
      reqQueue = mySession.findQueue(reqAtr);

      Thread.sleep(1000);

      //Create initial request.
      nEventProperties req = new nEventProperties();
      nConsumeEvent resp = new nConsumeEvent(req, "Request".getBytes());

      setQueueReader(respQueue);

      if (async) {
        System.out.println("Beginning to listen asynchronously...");
      } else {
        //Set up a thread to process incoming synchronous events.
        System.out.println("Beginning to listen synchronously...");
        Thread reader = new Thread(this);
        reader.setDaemon(true);
        reader.start();
      }

      //Construct transactional set of events if necessary(only one message in this example however).
      if (transactional) {
        System.out.println("Transactional");
        Vector<nConsumeEvent> vEvents = new Vector<nConsumeEvent>();
        nTransactionAttributes TXAttrib = new nTransactionAttributes(reqQueue, 1000);
        new nTransactionFactory();
        nTransaction tx = nTransactionFactory.create(TXAttrib);
        vEvents.addElement(resp);
        tx.publish(vEvents);
        tx.commit();
      } else {
        //Otherwise simply publish the event.
        reqQueue.push(resp);
      }
      System.out.println("Published request \"Request\"");
    } catch (Exception ex) {
      ex.printStackTrace();
    }

    //Exit on user input.
    try {
      System.in.read();
    } catch (IOException e) {
    }

    //Destroy the reader.
    try {
      nQueue.destroyReader(queueReader);
    } catch (Exception ex) {
      System.out.println("Already disconected.");
      ex.printStackTrace();
    }
    System.out.println("Finished");
    System.exit(0);

  }

  // Set the appropriate type of queue listener.
  private void setQueueReader(nQueue respQueue) {
    try {
      if (async) {
        if (transactional) {
          queueReader = respQueue.createAsyncTransactionalReader(new nQueueReaderContext(this));
        } else {
          queueReader = respQueue.createAsyncReader(new nQueueReaderContext(this));
        }
      } else {
        if (transactional) {
          queueReader = respQueue.createTransactionalReader(new nQueueReaderContext(this));
        } else {
          queueReader = respQueue.createReader(new nQueueReaderContext(this));
        }
      }
    } catch (Exception ex) {
      System.out.println("Could not create reader");
      System.exit(1);
    }
  }

  //Construct a session with the user name supplied.
  @Override
  protected void constructSession(String[] realmDetails) {

    // Create a realm session attributes object from the array of strings
    try {
      nsa = new nSessionAttributes(realmDetails, 2);
    } catch (Exception ex) {
      System.out.println("Error creating Session Attributes. Please check your RNAME");
      System.exit(1);
    }

    // Add this class as an asynchronous exception listener
    try {
      // Create a session object from the session attributes object,
      // passing this
      // as a reconnect handler class (optional). This will ensure that
      // the reconnection
      // methods will get called by the API.
      mySession = nSessionFactory.create(nsa, this, username);
      mySession.addAsyncExceptionListener(this);
      mySession.enableThreading(4);
    } catch (nIllegalArgumentException ex) {
    }

    // Initialise the Nirvana session. This physically opens the connection
    // to the
    // Nirvana realm, using the specified protocols. If multiple interfaces
    // are supported
    // these will be attempted in weight order (SSL, HTTPS, socket, HTTP).
    try {
      mySession.init();
      mySession.updateConnectionListWithServerList();
    }
    // Handle errors
    catch (nSecurityException sec) {
      System.out.println("The current user is not authorised to connect to the specified Realm Server");
      System.out.println("Please check the realm acls or contact support");
      sec.printStackTrace();
      System.exit(1);
    } catch (nRealmUnreachableException rue) {
      System.out.println("The Nirvana Realm specified by the RNAME value is not reachable.");
      System.out.println("Please ensure the Realm is running and check your RNAME value.");
      rue.printStackTrace();
      System.exit(1);
    } catch (nSessionNotConnectedException snce) {
      System.out.println("The session object used is not physically connected to the Nirvana Realm.");
      System.out.println("Please ensure the Realm is up and check your RNAME value.");
      snce.printStackTrace();
      System.exit(1);
    } catch (nSessionAlreadyInitialisedException ex) {
      System.out.println("The session object has already been initialised.");
      System.out.println("Please make only one call to the .init() function.");
      ex.printStackTrace();
      System.exit(1);
    }
  }

  public void go(nConsumeEvent evt) {
    //Deal with the response.
    System.out.println("Recieved Response :\"" + new String(evt.getEventData()) + "\"");
  }

  public void run() {
    //Deal with synchronous events i.e. the server response.
    System.out.println("Running Thread");
    nConsumeEvent evt;
    while (true) {
      try {
        if (transactional) {
          evt = ((nQueueSyncTransactionReader) queueReader).pop(-1);
        } else {
          evt = ((nQueueSyncReader) queueReader).pop(-1);
        }
        if (evt != null) {
          go(evt);
        }
      } catch (Exception e) {
        System.out.println("Exception in pop....exiting!");
        e.printStackTrace();
        break;
      }
    }
    System.exit(1);
  }
}