simplepublisher.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

              }
              System.out.println("Connection Factory ready");

              // Create a connection with the ConnectionFactory
              TopicConnection connection = factory.createTopicConnection();
              connection.start();
              System.out.println("Connection ready");

              // Create a session within the connection
              TopicSession session = connection.createTopicSession(transacted,ackType);
              System.out.println("Session ready");

              // Looking for / Creating a topic
              Topic topic = null;
              try
              {
                  topic = session.createTopic(topicName);
                  //topic = (Topic)context.lookup(topicName);
              }
              catch (Exception e)
              {
                  System.err.println("Failed to create Topic");
              }
              System.out.println("Topic - " + topicName + " ready");


              // Creating publisher
              TopicPublisher publisher = session.createPublisher(topic);
              System.out.println("Publisher ready");


              System.out.println("Publishing Messages ...");


              // Messages preparing

              // Messages Type
              TextMessage tMessage=null;     // Using simple text messages
              ObjectMessage oMessage=null;   // Using Lumos Alarm

              if (textMessages)
              {
                  // Setting the JMS Message
                  tMessage = session.createTextMessage();
                  tMessage.setText("JMS Text Message");
				  System.out.println("Size of Text : " + size("JMS Text Message"));
				  System.out.println("Size of JMS Text Message : " + size((Serializable)tMessage));				  
              }
              else
              {
                  // Defining Message structure variables
                  AlarmImpl mAlarm;
				  String mEntity = "Company" + "Lumos" + "Dept." + "R+D" + "User" + "Fabien" + "IP" + "10.0.0.137";
 				  String addData = "Vive la France ";
                  if (heavyweight)
                  {
                      addData = new String(new byte[weight]);
                      System.out.println("Heavyweight object building, size of addData : " + addData.length());

                  }
                  ArrayList mList = new ArrayList();
                  mList.add(0,"Item1");

                  // Creating an Alarm
                  mAlarm = new AlarmImpl(   // int id,
                                            0,
                                             // String name,
                                            "JMSTest",
                                            // EntityId src,
                                            mEntity,
                                            // String ct, - Condition Type
                                            "Message",
                                            // AlarmType type,
                                            "UNKNOWN",
                                            // AlarmSeverity sev,
                                            "WARNING",
                                            // long creTime,
                                            System.currentTimeMillis(),
                                            // boolean isSA, - Service Affecting ?
                                            true,
                                            // boolean isCAuto, - Auto Cleared
                                            false,
                                            // String add, - Additional Data
                                            addData,
                                            // List ad,
                                            mList,
                                            // boolean ack,
                                            true,
                                            // boolean isC - isCleared
                                            false
                                            );

                  // Setting the JMS Message
                  oMessage = session.createObjectMessage();
                  oMessage.setObject(mAlarm);
				  System.out.println("Size of Alarm : " + size(mAlarm));
				  System.out.println("Size of JMS Object Message : " + size((Serializable)oMessage));				                    

              }

              // Sending Messages
              long startTime = System.currentTimeMillis();
              if (textMessages)
              {
                  do
                  {
                      for (int i=0;i<count;i++)
                      {
                          absCount++;
                          publisher.publish(tMessage, persistency, 1, Long.MAX_VALUE);
                          //System.out.print("\rmsg: " + absCount);
                          //Thread.sleep(sleepTime);
                      }
                      System.out.print("\rmsg: " + absCount);
                      
                      if (infiniteCount)
                      {
                          // Mesurements
                          long endTime = System.currentTimeMillis();
                          Date startDate = new Date(startTime);
                          Date endDate = new Date(endTime);
                          long duration_ms = endTime - startTime;
                          double duration_s = duration_ms/1000;
                          double msgRate = absCount/duration_s;
                          System.out.println();
                          System.out.println("Started sending at : " + startDate
                                              + "\nCount at : " + endDate
                                              + "\nDuration : " + duration_s + " sec."
                                              + "\nMessage count : " + absCount
                                              + "\nMessage sending rate : " + msgRate + " msg/s.\n");
                      }
                  }
                  while (infiniteCount);
              }
              else
              {
                  do
                  {
                      for (int i=0;i<count;i++)
                      {
                          absCount++;
                          publisher.publish(oMessage, persistency, 1, Long.MAX_VALUE);
                          //System.out.print("\rmsg: " + absCount);
                          //Thread.sleep(sleepTime);
                      }
                      System.out.print("\rmsg: " + absCount);
                      
                      if (infiniteCount)
                      {
                          // Mesurements
                          long endTime = System.currentTimeMillis();
                          Date startDate = new Date(startTime);
                          Date endDate = new Date(endTime);
                          long duration_ms = endTime - startTime;
                          double duration_s = duration_ms/1000;
                          double msgRate = absCount/duration_s;
                          System.out.println();
                          System.out.println("Started sending at : " + startDate
                                              + "\nCount at : " + endDate
                                              + "\nDuration : " + duration_s + " sec."
                                              + "\nMessage count : " + absCount
                                              + "\nMessage sending rate : " + msgRate + " msg/s.\n");
                      }
                  }
                  while (infiniteCount);
              }
              long endTime = System.currentTimeMillis();

              // Mesurements
              Date startDate = new Date(startTime);
              Date endDate = new Date(endTime);
              long duration_ms = endTime - startTime;
              double duration_s = duration_ms/1000;
              double msgRate = count/duration_s;
              System.out.println();
              System.out.println("Started sending at : " + startDate
                                  + "\nEnded sending at : " + endDate
                                  + "\nSent Duration : " + duration_s + " sec."
                                  + "\nMessage count : " + count
                                  + "\nMessage sending rate : " + msgRate + " msg/s.");


              // Closing communication objects
              publisher.close();
              session.close();
              connection.close();

        }
        catch (Exception exception)
        {
            exception.printStackTrace();
        }
        finally
        {
            // run the gc and the finalizer
            System.gc();
            System.runFinalization();
            System.exit(0);
        }

    }

	/**
	*
	*/ 
	private static int size( Serializable x)
	{
		try
		{
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream( baos);
			oos.writeObject( x);
			oos.close();
			baos.close();
			return baos.toByteArray().length;
		}
		catch( IOException io)
		{
			io.printStackTrace();
			return -1;
		}
	}

    static protected void usage()
    {
		PrintStream out = System.out;

		out.println("\n\n");
		out.println("==================================================================================");
		out.println("        Usage information for openjms.examples.profiling.SimplePublisher");
		out.println("==================================================================================");
		out.println();
		out.println("openjms.examples.profiling.SimplePublisher");
		out.println();
		out.println("    [-help | -jndiport <port number> | -jndihost <host address> | -ack <auto/client/dups>");
		out.println("     -count <msg nb> | -infinite | -heavyweight <size_bytes> | -persistent | -text");
                out.println("     -topic <topic name> | -transacted | -test]");
		out.println();
		out.println("Misc. options :");
		out.println("\t-help         displays this screen");
		out.println("\t-test         displays command line argument values");
		out.println();
		out.println("Administered objects options :");
		out.println("\t-jndiport     port where the jndi server runs    - Defaut = 1099");
		out.println("\t-jndihost     host where jndi server runs        - Defaut = localhost");
		out.println();
		out.println("JMS Messaging options :");
		out.println("\t-ack          \"auto\", \"client\" or \"dups\"        - Default = auto");
		out.println("\t-count        number of message to be sent            - Default = 500");
		out.println("\t-inifite      infinite number of message to be sent   - Default = false");
		out.println("\t-heavyweight  size of the alarm to send               - Default = false");
		out.println("\t-persistent   make the messages persistent            - Default = false");
		out.println("\t-text         only text messages to be sent           - Default = false");
		out.println("\t-topic        topic name to subscriber under.         - Default = topic");
		out.println("\t-transacted   use transacted session with the Topic.  - Default = false");
                out.println("\n\n");
                System.exit(0);
    }


}