package javacodebook.xml.transport.jms.p2p;

import javax.jms.*;
import javax.naming.*;

/**
 * Die XMLQueueReceiver-Klasse benutzt eine Session, um XML-Dokumente in Form
 * von TextMessages von einer Queue zu lesen. Dabei werden XML-Dokumente als
 * reine Text-Dokumente behandelt. Das Parsen und eventuelle Validierung muss getrennt
 * geschehen.
 */
public class XMLQueueReceiver {

  private static final String USAGE="\nBenutzerhinweis: javacodebook.xml.transport.jms.p2p.XMLQueueReceiver "+
          "<laufzeitSekunden> <warteSchlangenNamen>\n\nwobei\n\n<laufzeitSekunden>\ndie Anzahl "+
          "der Sekunden ist, die Nachrichten empfangen werden sollen und \n\n<warteSchlangenNamen>\n"+
          "der Warteschlangenname ist, von der die Nachrichten empfangen werden sollen";
  private long end;

  /**
   * An die main-Methode müssen die Parameter <laufzeitSekunden> und <warteschlangennamen> übergeben werden.
   * In der Methode wird ein XMLQueueReceiver-Objekt instanziiert, es werden die Kommandozeilen-Parameter ausgelesen und die
   * receiveMessages()-Methode mit den beiden ausgelesenen Parametern <laufzeitSekunden> und <warteschlangennamen>
   * aufgerufen.
   */
  public static void main(String args[])
  {
    XMLQueueReceiver xMLQueueReceiver=new XMLQueueReceiver();
    long timeToReceiveMessages=0;
    String queueName="not set";

    if(args.length!=2)
    {
      System.out.println(getUsage());
      System.exit(1);
    }
    else
    {
      try
      {
        timeToReceiveMessages=new Long(args[0]).longValue();
      }
      catch(NumberFormatException e)
      {
        System.out.println("\nBenutzerhinweis:  Bitte als erstes Argument eine Zahl eingeben");
        System.exit(1);
      }
      queueName=args[1];
    }
    xMLQueueReceiver.receiveMessages(queueName, timeToReceiveMessages);
  }

  public void receiveMessages(String queueName, long timeToReceiveMessages) {

    Context                 jndiContext = null;
    QueueConnectionFactory  queueConnectionFactory = null;
    QueueConnection         queueConnection = null;
    QueueSession            queueSession = null;
    Queue                   queue = null;
    QueueReceiver           queueReceiver = null;

    setRuntime(timeToReceiveMessages);

    System.out.println("Queue name is " + queueName);

    // Erzeugung eines neuen JNDI API InitialContext Objektes
    try {
      jndiContext = new InitialContext();
    } catch (NamingException e) {
      System.out.println("Es konnte kein JNDI API Kontext erstellt werden: "+e.toString());
      System.exit(1);
    }

    // Look-up der Connection-Factory und der Queue.  Falls eines der Objekte nicht
    // gefunden wird, soll die Anwendung verlassen werden.
    try {
      queueConnectionFactory = (QueueConnectionFactory)jndiContext.lookup("QueueConnectionFactory");
      queue = (Queue) jndiContext.lookup(queueName);
    } catch (NamingException e) {
      System.out.println("JNDI API lookup verfehlt: " +
                         e.toString()+"\n\nentweder die QueueconnectionFactory oder die Warteschlange namens "+
                         queueName+" ist nicht beim Namensdienst registriert");
      System.exit(1);
    }

    try {

      // Eine neue Connection wird erzeugt.
      queueConnection = queueConnectionFactory.createQueueConnection();

      // Über die Connection wird ein QueueSession-Objekt erzeugt. Dadurch dass true übergeben wird,
      // ist die Verbindung transaktional - bei false wäre sie das nicht. Als 2. Parameter wir 0 übergeben um
      // anzudeuten, dass er bei transaktionalen QueueSessions keine Rolle spielt.  Falls keine transaktionale
      // QueueSession erzeugt wird, muss der 2. Parameter einer der Werte:
      //  * Session.AUTO_ACKNOWLEDGE;
      //  * Session.CLIENT_ACKNOWLEDGE;
      //  * Session.DUPS_OK_ACKNOWLEDGE;
      // sein.
      queueSession = queueConnection.createQueueSession(true, 0);

      // Erzeugung eines QueueReceiver-Objektes über das QueueSession-Objekt
      queueReceiver = queueSession.createReceiver(queue);

      // Bevor die Connection benutzt werden kann, muss sie gestartet werden
      queueConnection.start();

      // Beginn der Nachrichtenübertragung.
      String xmlDocument=null;
      while (System.currentTimeMillis()<end) {

        // Der Receiver empfängt während seiner Laufzeit Nachrichten von der Warteschlage.
        // Falls schon eine Nachricht in der Warteschlange ist, wird diese genommen.  Ansonsten wird in der
        // receive()-Methode genau 1 Millisekunde gewartet bis eine Nachricht in die Schlange gestellt wird.
        // Danach wird in die nächste Interation gegangen.
        Message message = queueReceiver.receive(1);
        if (message != null) {
          if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;

            // Zwischenspeicherung des empfangenen XML-Dokumentes in einer lokalen Variablen.
            // Die Weiterverarbeitung darf erst nach dem Aufruf der commit()-Methode auf das QueueSession-Objekt
            // stattfinden.
            xmlDocument=textMessage.getText();
          }
        }
        else
        {
          xmlDocument=null;
        }

        // Da die QueueSession transaktional ist, muss unbedingt commit()-Methode aufgerufen werden.
        // Falls das nicht geschieht, geht der JMS-Provider davon aus, dass die Nachricht nocht nicht
        // richtig ausgeliefert wurde und wird die Nachricht so lange in seiner Warteschlange aufbewahren, selbst
        // wenn er neu gestartet würde, bis ein Empfänger die Nachricht abholt und über die commit()-Methode
        // den ordnungsgemäßen Empfang quititert.
        try
        {
          queueSession.commit();

          // Verarbeitung der Nachricht:
          // Schreibe empfangene Nachricht in die Standardausgabe.  Hier käme im
          // Normalfall die Weiterverarbeitung des empfangenen XMLs ins Spiel --
          // also das Parsen samt entsprechender Applikationslogik.
          if(xmlDocument!=null) System.out.println(xmlDocument);
        }

        // Falls die Transaktion von Seiten des JMS-Providers nicht 'commited' werden konnte, wird eine JMSException geworfen, die
        // auch vom Typ der Unterklassen TransactionRolledBackException oder IllegalStateException sein kann.
        // Es empfiehlt sich, in solchen Fällen die empfangene Nachricht nicht weiter zu verarbeiten, da sie von dem Provider noch
        // mal ausgeliefert werden wird und somit eine doppelte Verarbeitung stattfinden würde.
        catch(JMSException e)
        {
          System.out.println("Ausnahmefall eingetreten:  "+e);
        }
      }
    } catch (JMSException e) {
      System.out.println("Ausnahmefall eingetreten: "+e.toString());
    } finally {
      if (queueConnection != null) {
        try {

          // Die Connection wird geschlossen.
          queueConnection.close();
          } catch (JMSException e) {}
      }
    }
  }

  public static String getUsage()
  {
    return USAGE;
  }

  /**
   * Mit dieser Methode wir das end-Attribut gesetzt.
   */
  private void setRuntime(long timeToReceiveMessages)
  {
    end=System.currentTimeMillis()+timeToReceiveMessages*1000;
  }
}

--- Neue Klasse ---

package javacodebook.xml.transport.jms.p2p;

import javax.jms.*;
import javax.naming.*;
import java.io.*;

/**
 * Die XMLQueueSender-Klasse benutzt ein QueueSession-Objekt, um XML-Dokumente in Form
 * von TextMessages in eine Queue zu schreiben. Die XML-Dokumente werden dabei
 * von dem Dateisystem gelesen und als einfache Strings behandelt.
 * Das Parsen und eventuelle Validierung muss getrennt geschehen.
 */
public class XMLQueueSender {

  private static final String USAGE="\nBenutzerhinweis: javacodebook.xml.transport.jms.p2p.XMLQueueSender "+
                                    "<warteSchlangenNamen> <dateiName> [<dateiName> ...]\n\nwobei\n\n<warteSchlangenNamen>\n"+
                                    "der Name der Warteschlange ist und\n\n<dateiName> [<dateiName> ...]\n"+
                                    "ein odere mehrer durch Leerzeichen getrennte Dateinamen von XML-Dateien sind, die über"+
                                    " die Warteschlange verschickt werden sollen";
  /**
   * An die main-Methode müssen die Parameter <warteschlangennamen> und eine durch Leerzeichen getrennte Folge von
   * Dateinamen <dateiName> [<dateiName> ...] übergeben werden.  Es wird ein Objekt vom Typ XMLQueueSender instanziiert und die
   * Kommandozeilenparameter ausgelesen. Nun wird  auf das XMLQueueSender-Objekt die Methode sendDocuments()
   * aufgerufen, wobei der erste Parameter als Warteschlangenname übergeben wird und die restlichen Parameter
   * in Form eines String-Arrays von XML-Dateinamen übergeben werden.
   */
  public static void main(String args[])
  {
    XMLQueueSender xMLQueueSender=new XMLQueueSender();

    if(args.length<2)
    {
      System.out.println(getUsage());
      System.exit(1);
    }
    else
    {
      String queueName=args[0];

      //Ein neuer String-Array für die XML-Dateinamen wird angelegt.
      String[] fileNames=new String[args.length-1];
      for(int i=1; i<args.length; i++)
      {
        fileNames[i-1]=args[i];
      }

      // queueName und die XML-Dateinamen in Form eines String-Arrays werden and die sendDocuments-Methode übergeben.
      xMLQueueSender.sendDocuments(queueName,fileNames);
    }
  }

  public void sendDocuments(String queueName, String[] fileNames) {

    Context                 jndiContext = null;
    QueueConnectionFactory  queueConnectionFactory = null;
    QueueConnection         queueConnection = null;
    QueueSession            queueSession = null;
    Queue                   queue = null;
    QueueSender             queueSender = null;

    System.out.println("Der Warteschlangenname ist " + queueName);

    // Erzeugung eines neuen JNDI API InitialContext Objektes
    try {
      jndiContext = new InitialContext();
    } catch (NamingException e) {
      System.out.println("Es konnte kein JNDI API Kontext erstellt werden: "+e.toString());
      System.exit(1);
    }

    // Look-up der Connection-Factory und der Queue. Falls eines der Objekte nicht
    // gefunden wird, soll die Anwendung verlassen werden.
    try {
      queueConnectionFactory = (QueueConnectionFactory)jndiContext.lookup("QueueConnectionFactory");
      queue = (Queue) jndiContext.lookup(queueName);
    } catch (NamingException e) {
      System.out.println("JNDI API lookup verfehlt: " +
                         e.toString()+"\n\nentweder die QueueConnectionFactory oder die Warteschlange namens "+
                         queueName+" ist nicht beim Namensdienst registriert");
      System.exit(1);
    }

    try {
      // Eine neue Connection wird erzeugt.
      queueConnection = queueConnectionFactory.createQueueConnection();

      // Über die QueueConnection wird ein QueueSession-Objekt erzeugt. Dadurch dass true übergeben wird,
      // ist die Verbindung transaktional - bei false wäre sie das nicht. Als 2. Parameter wir 0 übergeben um
      // anzudeuten, dass er bei transaktionalen QueueSessions keine Rolle spielt. Falls keine transaktionale
      // QueueSession erzeugt wird, muss der 2. Parameter einer der Werte:
      //  * Session.AUTO_ACKNOWLEDGE;
      //  * Session.CLIENT_ACKNOWLEDGE;
      //  * Session.DUPS_OK_ACKNOWLEDGE;
      // sein.
      queueSession = queueConnection.createQueueSession(true,0);

      // Erzeugung eines QueueSender-Objektes über das QueueSession-Objekt
      queueSender = queueSession.createSender(queue);

      // Erzeugung einer TextMessage über das QueueSession-Objekt
      TextMessage message = queueSession.createTextMessage();

      // Für jedes der zu verschickenden XML-Dokumente wird das TextMessage-Objekt
      // mit dem Text aus der XML-Datei belegt und über die send()-Methode des QueueSession-Objektes
      // abgeschickt.
      for (int i = 0; i < fileNames.length; i++) {
        String document=loadDocument(fileNames[i]);
        if(document!=null)
        {
          message.setText(document);
          System.out.println("Sende Nachricht:\n"+message.getText());
          queueSender.send(message);
        }

        // Da das QueueSession-Objekt transaktional ist, können die Nachrichten,
        // die über dieses QueueSession-Objekt verschickt wurden, von keinem Empfänger gelesen
        // werden, bevor nicht die commit()-Methode auf das QueueSession-Objekt aufgerufen wurde.
        queueSession.commit();
      }
    } catch (JMSException e) {
      System.out.println("Ausnahmezustand aufgetreten: "+e.toString());
    } finally {
      if (queueConnection != null) {
        try {
          // Zum Schluss muss die QueueConnection geschlossen werden, um nicht unnötig Resourcen zu belegen.
          queueConnection.close();
          } catch (JMSException e) {}
      }
    }
  }

  /**
   * Diese Methode liest die Datei <fileName> und gibt dessen Inhalt als
   * String zurück.
   */
  private String loadDocument(String fileName)
  {
    InputStream is=null;
    String document="";
    try
    {
      is=new FileInputStream(fileName);
      BufferedReader br=new BufferedReader(new InputStreamReader(is));
      String line="";

      while (line!=null)
      {
        document=document+line;
        line=br.readLine();
      }
    }
    catch(Exception e)
    {
      System.out.println("Probleme beim lesen des Dokuments "+fileName+": "+e);
      return null;
    }
    return document;
  }

  public static String getUsage()
  {
    return USAGE;
  }
}