import com.solacesystems.jcsmp.JCSMPSession; //导入依赖的package包/类
public ProducerThread(String jobId, List<WorkInstance> workQueue,
		JCSMPSession session, Destination destForPublish, long msgRate,
		JobRequestStats jobStatus, GlobalStats globalStats) {
	this.jobId = jobId;
	this.workQueue = workQueue;
	this.session = session;
	this.destForPublish = destForPublish;
	this.msgRate = msgRate;
	this.jobStatus = jobStatus;
	this.globalStats = globalStats;
}
 

import com.solacesystems.jcsmp.JCSMPSession; //导入依赖的package包/类
public void run(String[] args) {

	try {

        System.out.println("OutputProxy initializing...");

        // Create a JCSMP Session
        JCSMPProperties properties = new JCSMPProperties();
        properties.setProperty(JCSMPProperties.HOST, args[0]); 		// msg-backbone ip:port
        properties.setProperty(JCSMPProperties.VPN_NAME, args[1]); 	// message-vpn
        properties.setProperty(JCSMPProperties.USERNAME, args[2]); 	// client-username (assumes no password)
	String t = args[3];
	Topic topic = JCSMPFactory.onlyInstance().createTopic(t);       // topic to subscribe to stream
	String host = args[4];						// redirect udp stream to this host
	port = Integer.parseInt(args[5]);				// redirect udp stream to this port	
	verbose = args[6];	   					// verbose
	
        session = JCSMPFactory.onlyInstance().createSession(properties);

        session.connect();

        Queue queue = session.createTemporaryQueue();

        FlowReceiver receiver = session.createFlow(queue, null, this);

        session.addSubscription(queue,topic,JCSMPSession.WAIT_FOR_CONFIRM);
        
        dsocket = new DatagramSocket();
        address = InetAddress.getByName(host);

        System.out.println("Connected.");
	System.out.println("Control-C to exit");

        receiver.start();
	Thread.sleep(1000000);

	} catch (Exception ex) {
            System.err.println("Encountered an Exception... " + ex.getMessage());
	}

    }
 

import com.solacesystems.jcsmp.JCSMPSession; //导入依赖的package包/类
@PostConstruct
public void init() {
	// Connect to Solace
	trace.info("************* Init Called ************");
	trace.info(System.getenv("VCAP_SERVICES"));

	CloudFactory cloudFactory = new CloudFactory();
	Cloud cloud = cloudFactory.getCloud();
	
	SolaceMessagingInfo solaceMessagingServiceInfo =
			(SolaceMessagingInfo) cloud.getServiceInfo("solace-messaging-demo-instance");
	
	if (solaceMessagingServiceInfo == null) {
		trace.error("Did not find instance of 'solace-messaging' service");
		trace.error("************* Aborting Solace initialization!! ************");
		return;
	}
	
	trace.info("Solace client initializing and using SolaceMessagingInfo: " + solaceMessagingServiceInfo);

	final JCSMPProperties properties = new JCSMPProperties();
	properties.setProperty(JCSMPProperties.HOST, solaceMessagingServiceInfo.getSmfHost());
	properties.setProperty(JCSMPProperties.VPN_NAME, solaceMessagingServiceInfo.getMsgVpnName());
	properties.setProperty(JCSMPProperties.USERNAME, solaceMessagingServiceInfo.getClientUsername());
	properties.setProperty(JCSMPProperties.PASSWORD, solaceMessagingServiceInfo.getClientPassword());
	
	try {
		session = JCSMPFactory.onlyInstance().createSession(properties);
		session.connect();
		
		clientName = (String)session.getProperty(JCSMPProperties.CLIENT_NAME);

		final Queue queue = JCSMPFactory.onlyInstance().createQueue("Q/demo/requests");
		 
		// set queue permissions to "consume" and access-type to "exclusive"
		final EndpointProperties endpointProps = new EndpointProperties();
		endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME);
		endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_NONEXCLUSIVE);
		 
		// Actually provision it, and do not fail if it already exists
		session.provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS);
		
		producer = session.getMessageProducer(new PublisherEventHandler());

		final ConsumerFlowProperties flowProp = new ConsumerFlowProperties();
		flowProp.setEndpoint(queue);
		flowProp.setAckMode(JCSMPProperties.SUPPORTED_MESSAGE_ACK_CLIENT);
		
		final FlowReceiver cons = session.createFlow(new DemoMessageListener(), flowProp, endpointProps);
		cons.start();
		
		System.out.println("************* Solace initialized correctly!! ************");
		

	} catch (Exception e) {
		Trace.error("Error connecting and setting up session.", e);
	}
}
 

import com.solacesystems.jcsmp.JCSMPSession; //导入依赖的package包/类
public void run(String... args) throws JCSMPException, InterruptedException {
    final LinkedList<MsgInfo> msgList = new LinkedList<MsgInfo>();

    System.out.println("ConfirmedPublish initializing...");
    // Create a JCSMP Session
    final JCSMPProperties properties = new JCSMPProperties();
    properties.setProperty(JCSMPProperties.HOST, args[0]);     // host:port
    properties.setProperty(JCSMPProperties.USERNAME, args[1].split("@")[0]); // client-username
    properties.setProperty(JCSMPProperties.PASSWORD, args[2]); // client-password
    properties.setProperty(JCSMPProperties.VPN_NAME,  args[1].split("@")[1]); // message-vpn
    final JCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);
    session.connect();

    String queueName = "Q/tutorial";
    final Queue queue = JCSMPFactory.onlyInstance().createQueue(queueName);

    /** Correlating event handler */
    final XMLMessageProducer prod = session.getMessageProducer(new PubCallback());

    // Publish-only session is now hooked up and running!
    System.out.printf("Connected. About to send " + count + " messages to queue '%s'...%n", queue.getName());

    for (int i = 1; i <= count; i++) {
        TextMessage msg = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
        msg.setDeliveryMode(DeliveryMode.PERSISTENT);
        String text = "Confirmed Publish Tutorial! Message ID: " + i;
        msg.setText(text);

        // The application will wait and confirm the message is published
        // successfully.
        // In this case, wrap the message in a MsgInfo instance, and
        // use it as a correlation key.
        final MsgInfo msgCorrelationInfo = new MsgInfo(i);
        msgCorrelationInfo.sessionIndependentMessage = msg;
        msgList.add(msgCorrelationInfo);

        // Set the message's correlation key. This reference
        // is used when calling back to responseReceivedEx().
        msg.setCorrelationKey(msgCorrelationInfo);

        // Send message directly to the queue
        prod.send(msg, queue);
    }
    System.out.println("Messages sent. Processing replies.");
    try {
        latch.await(); // block here until message received, and latch will flip
    } catch (InterruptedException e) {
        System.out.println("I was awoken while waiting");
    }

    // Process the replies
    while (msgList.peek() != null) {
        final MsgInfo ackedMsgInfo = msgList.poll();
        System.out.printf("Removing acknowledged message (%s) from application list.\n", ackedMsgInfo);
    }

    // Close session
    session.closeSession();
}