/* * * 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 com.pcbsys.foundation.utils.xmlHelper; import com.pcbsys.nirvana.client.*; import java.io.*; /** * Subscribes to a nirvana channel that contains XML messages */ public class XMLSubscribe extends nSampleApp implements nEventListener { static long starteid; static String selector = null; private long lastEID = 0; private long startTime = 0; private long byteCount = 0; private int logLevel = 0; private int count = -1; private int reportCount = 1000; private xmlHelper xh = null; private nChannel myChannel; private static XMLSubscribe mySelf = null; /** * This method demonstrates the Nirvana API calls necessary to subscribe to * a channel that contains XML messages. It is called after all command line * arguments have been received and validated * * @param realmDetails a String[] containing the possible RNAME values * @param achannelName the channel name to create * @param selector the subscription selector filter * @param startEid the eid to start subscribing from * @param loglvl the specified log level * @param repCount the specified report count */ private void doit(String[] realmDetails, String achannelName, String selector, int startEid, int loglvl, int repCount) { logLevel = loglvl; reportCount = repCount; mySelf.constructSession(realmDetails); // Subscribes to the specified channel try { // Create a channel attributes object nChannelAttributes nca = new nChannelAttributes(); nca.setName(achannelName); // Obtain the channel reference myChannel = mySession.findChannel(nca); // if the latest event has been implied (by specifying -1) if (startEid == -1) { // Get the last eid on the channel and reset the start eid with // that value starteid = myChannel.getLastEID(); } // Add this object as a subscribe to the channel with the specified // message selector // and start eid myChannel.addSubscriber(this, selector, starteid); // Stay subscribed until the user presses any key System.out.println("Press any key to quit !"); BufferedInputStream bis = new BufferedInputStream(System.in); try { bis.read(); } catch (Exception read) { } // Ignore this // Remove this subscriber myChannel.removeSubscriber(this); } // Handle errors catch (nChannelNotFoundException cnfe) { System.out.println("The channel specified could not be found."); System.out.println("Please ensure that the channel exists in the REALM you connect to."); cnfe.printStackTrace(); System.exit(1); } catch (nSecurityException se) { System.out.println("Insufficient permissions for the requested operation."); System.out.println("Please check the ACL settings on the server."); se.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 (nUnexpectedResponseException ure) { System.out.println("The Nirvana REALM has returned an unexpected response."); System.out.println("Please ensure the Nirvana REALM and client API used are compatible."); ure.printStackTrace(); System.exit(1); } catch (nUnknownRemoteRealmException urre) { System.out.println("The channel specified resided in a remote realm which could not be found."); System.out.println("Please ensure the channel name specified is correct."); urre.printStackTrace(); System.exit(1); } catch (nRequestTimedOutException rtoe) { System.out.println("The requested operation has timed out waiting for a response from the REALM."); System.out.println("If this is a very busy REALM ask your administrator to increase the client timeout values."); rtoe.printStackTrace(); System.exit(1); } catch (nChannelAlreadySubscribedException chase) { System.out.println("You are already subscribed to this channel."); chase.printStackTrace(); System.exit(1); } catch (nSelectorParserException spe) { System.out.println("An error occured while parsing the selector filter specified."); System.out.println("Please check the JMS documentation on how to write a valid selector."); spe.printStackTrace(); System.exit(1); } catch (nBaseClientException nbce) { System.out.println("An error occured while creating the Channel Attributes object."); nbce.printStackTrace(); System.exit(1); } // Close the session we opened try { nSessionFactory.close(mySession); } catch (Exception ex) { } // Close any other sessions within this JVM so that we can exit nSessionFactory.shutdown(); } protected void processArgs(String[] args) { switch (args.length) { case 5: System.getProperties().put("SELECTOR", args[4]); case 4: System.getProperties().put("COUNT", args[3]); case 3: System.getProperties().put("DEBUG", args[2]); case 2: System.getProperties().put("START", args[1]); case 1: if (args[0].equals("-?")) { UsageEnv(); } System.getProperties().put("CHANNAME", args[0]); break; } } public static void main(String[] args) { // Create an instance for this class mySelf = new XMLSubscribe(); // Process command line arguments mySelf.processArgs(args); // Process Environment Variables nSampleApp.processEnvironmentVariables(); // Check the channel name specified String channelName = null; if (System.getProperty("CHANNAME") != null) { channelName = System.getProperty("CHANNAME"); } else { Usage(); System.exit(1); } int startEid = -1; // Default value (points to last event in the channel // + 1) // Check to see if a start EID value has been specified if (System.getProperty("START") != null) { try { startEid = Integer.parseInt(System.getProperty("START")); } catch (Exception num) { } // Ignore and use the defaults } int loglvl = 1; // Default value // Check to see if a DEBUG value has been specified. if (System.getProperty("DEBUG") != null) { try { loglvl = Integer.parseInt(System.getProperty("DEBUG")); } catch (Exception num) { } // Ignore and use the default } int reportCount = 1000; // Default value // Check to see if a value for report count has been specified. Every // time // N events get received where N = report count, a subscription rate // report will // be printed in System.out if (System.getProperty("COUNT") != null) { try { reportCount = Integer.parseInt(System.getProperty("COUNT")); } catch (Exception num) { } // Ignore and use the default } // Check for a selector message filter value String selector = System.getProperty("SELECTOR"); // Check the local realm details String RNAME = null; if (System.getProperty("RNAME") != null) { RNAME = System.getProperty("RNAME"); } else { Usage(); System.exit(1); } // Process the local REALM RNAME details String[] rproperties = new String[4]; rproperties = parseRealmProperties(RNAME); // Subscribe to the channel specified mySelf.doit(rproperties, channelName, selector, startEid, loglvl, reportCount); } /** * A callback is received by the API to this method each time an event is * received from the nirvana channel. Be carefull not to spend too much time * processing the message inside this method, as until it exits the next * message can not be pushed. * * @param evt An nConsumeEvent object containing the message received from * the channel */ public void go(nConsumeEvent evt) { // If this is the first message we receive if (count == -1) { // Get a timestamp to be used for message rate calculations startTime = System.currentTimeMillis(); count = 0; } // Increment he counter count++; // Have we reached the point where we need to report the rates? if (count == reportCount) { // Reset the counter count = 0; // Get a timestampt to calculate the rates long end = System.currentTimeMillis(); // Does the specified log level permits us to print on the screen? if (logLevel >= 1) { // Dump the rates on the screen System.out.println( "Received " + reportCount + " in " + (end - startTime) + " Evt/Sec = " + ((reportCount * 1000) / (end - startTime)) + " Bytes/sec=" + ((byteCount * 1000) / (end - startTime))); } // Set the startTime for the next report equal to the end timestamp // for the previous one startTime = end; // Reset the byte counter byteCount = 0; } // If the last EID counter is not equal to the current event ID if (lastEID != evt.getEventID()) { // If yes, maybe we have missed an event, so print a message on the // screen. // This message could be printed for a number of other reasons. // One of them would be someone purging a range creating an 'eid // gap'. // As eids are never reused within a channel you could have a // situation // where this gets printed but nothing is missed. System.out.println("Expired event range " + (lastEID) + " - " + (evt.getEventID() - 1)); // Reset the last eid counter lastEID = evt.getEventID() + 1; } else { // Increment the last eid counter lastEID++; } // Get the data of the message byte[] buffer = evt.getEventData(); if (buffer != null) { // Add its length to the byte counter byteCount += buffer.length; } // If the loglevel permits printing on the screen if (logLevel >= 2) { // Print the eid System.out.println("Event id : " + evt.getEventID()); // If the loglevel permits printing on the screen if (logLevel >= 3) { // Print the message tag System.out.println("Event tag : " + evt.getEventTag()); // Print the message data try { System.out.print("Event data : "); if (xh == null) { xh = new xmlHelper(evt.getDocument()); } else { xh.setDoc(evt.getDocument()); } xh.dumpDoc(); } catch (Exception xmlError) { System.out.println("\nAn error occured while parsing the XML data contained in the event"); } if (evt.hasAttributes()) { displayEventAttributes(evt.getAttributes()); } nEventProperties prop = evt.getProperties(); if (prop != null) { displayEventProperties(prop); } } } } /** * Prints the usage message for this class */ private static void Usage() { System.out.println("Usage ...\n"); System.out.println("nxmlsub [start eid] [debug] [count] [selector] \n"); System.out.println(" \n"); System.out.println(" - Channel name parameter for the channel to subscribe to"); System.out.println("\n[Optional Arguments] \n"); System.out.println("[start eid] - The Event ID to start subscribing from"); System.out.println("[debug] - The level of output from each event, 0 - none, 1 - summary, 2 - EIDs, 3 - All"); System.out.println("[count] - The number of events to wait before printing out summary information"); System.out.println("[selector] - The event filter string to use"); System.out.println("\n\nNote: -? provides help on environment variables \n"); } } // End of XMLSubscribe Class