使用Java进行CORBA编程-JacORB-Notification Service(2)

上一篇介绍了CORBA的NotificationService以及如何来使用NotificationService。
上一篇 使用Java进行CORBA编程-JacORB-Notification Service

下面提供使用JacORB的NotificatoinService的一个例程，运行此例子需要启动NameService，
启动方法和例程的启动方法见 使用Java进行CORBA编程-JacORB-入门

另外，还需要启动NotificatoinService，在Windows上可以到JacORB的安装目录下的bin，使用下面的脚本来启动。
需要指定NotificationService使用的端口号，这里是7981。
注意，如果启动中出现错误，可能需要修改JacORB的安装目录下的etc/jacorb.properties文件中的某些设置。

set CLASSPATH=.
set CLASSPATH=%CLASSPATH%;../lib/slf4j-api-1.6.4.jar
set CLASSPATH=%CLASSPATH%;../lib/slf4j-jdk14-1.6.4.jar
set CLASSPATH=%CLASSPATH%;../lib/jacorb.jar
set CLASSPATH=%CLASSPATH%;../lib/jacorb-services.jar
set CLASSPATH=%CLASSPATH%;../lib/antlr-2.7.2.jar
set CLASSPATH=%CLASSPATH%;../lib/picocontainer-1.2.jar

set SYSTEM_PROPS=-Djacorb.home=..
set SYSTEM_PROPS=%SYSTEM_PROPS% -Dorg.omg.CORBA.ORBClass=org.jacorb.orb.ORB
set SYSTEM_PROPS=%SYSTEM_PROPS% -Dorg.omg.CORBA.ORBSingletonClass=org.jacorb.orb.ORBSingleton
java %SYSTEM_PROPS% -cp %CLASSPATH% org.jacorb.notification.ConsoleMain -port 7981 -printCorbaloc

Server代码
1）StructuredPushSupplierImpl.java

package learning.corba.notification.newsbroadcast.server;

import org.omg.CosNotification.EventType;
import org.omg.CosNotifyComm.InvalidEventType;
import org.omg.CosNotifyComm.StructuredPushSupplierPOA;

public class StructuredPushSupplierImpl extends StructuredPushSupplierPOA {

	public void disconnect_structured_push_supplier() {
		System.out.println("disconnect_structured_push_supplier");
	}

	public void subscription_change(EventType[] added, EventType[] removed)
			throws InvalidEventType {
		throw new UnsupportedOperationException();
	}

}

2）NewsBCNotifServer.java

package learning.corba.notification.newsbroadcast.server;

import java.util.Properties;
import java.util.Random;

import org.omg.CORBA.IntHolder;
import org.omg.CORBA.ORB;
import org.omg.CosEventComm.Disconnected;
import org.omg.CosNaming.NamingContextExt;
import org.omg.CosNaming.NamingContextExtHelper;
import org.omg.CosNotification.DefaultPriority;
import org.omg.CosNotification.EventHeader;
import org.omg.CosNotification.EventType;
import org.omg.CosNotification.FixedEventHeader;
import org.omg.CosNotification.Priority;
import org.omg.CosNotification.Property;
import org.omg.CosNotification.StructuredEvent;
import org.omg.CosNotifyChannelAdmin.ClientType;
import org.omg.CosNotifyChannelAdmin.EventChannel;
import org.omg.CosNotifyChannelAdmin.EventChannelFactory;
import org.omg.CosNotifyChannelAdmin.EventChannelFactoryHelper;
import org.omg.CosNotifyChannelAdmin.StructuredProxyPushConsumer;
import org.omg.CosNotifyChannelAdmin.StructuredProxyPushConsumerHelper;
import org.omg.CosNotifyChannelAdmin.SupplierAdmin;
import org.omg.CosNotifyComm.InvalidEventType;
import org.omg.CosNotifyComm.StructuredPushSupplierHelper;
import org.omg.PortableServer.POA;
import org.omg.PortableServer.POAHelper;

public class NewsBCNotifServer implements Runnable {

	private ORB orb;
	private StructuredProxyPushConsumer structuredProxyPushConsumer;

	public static void main(String[] args) {
		try {
			Properties props = new Properties();
			props.put("org.omg.PortableInterceptor.ORBInitializerClass.bidir_init",
					  "org.jacorb.orb.giop.BiDirConnectionInitializer");
			
			ORB orb = ORB.init(args, props);
			POA rootPoa = POAHelper.narrow(
					orb.resolve_initial_references("RootPOA"));
			rootPoa.the_POAManager().activate();
				
			NamingContextExt ncExt = NamingContextExtHelper.narrow(
					orb.resolve_initial_references("NameService"));
			
			EventChannelFactory eventChannelFactory = EventChannelFactoryHelper.narrow(
					orb.resolve_initial_references("NotificationService"));
			
            		Property[] initialQos = new Property[0];
            		Property[] initialAdmin = new Property[0];
			EventChannel eventChannel = eventChannelFactory.create_channel(
					initialQos, initialAdmin, new IntHolder());
			
			ncExt.rebind(ncExt.to_name("NewsBC.NotifChannel"), eventChannel);
			
			StructuredPushSupplierImpl pushSupplierImpl = new StructuredPushSupplierImpl();
			SupplierAdmin supplierAdmin = eventChannel.default_supplier_admin();
			StructuredProxyPushConsumer proxyPushConsumer =
				StructuredProxyPushConsumerHelper.narrow(
					supplierAdmin.obtain_notification_push_consumer(
					ClientType.STRUCTURED_EVENT, new IntHolder()));
			proxyPushConsumer.connect_structured_push_supplier(
					StructuredPushSupplierHelper.narrow(
							rootPoa.servant_to_reference(pushSupplierImpl)));
			
			Thread t = new Thread(new NewsBCNotifServer(orb, proxyPushConsumer));
			t.start();
			
			orb.run();
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public NewsBCNotifServer(ORB orb, StructuredProxyPushConsumer proxyPushConsumer) {
		this.orb = orb;
		this.structuredProxyPushConsumer = proxyPushConsumer;
	}
	
	public void run() {
		int i = 0;
		
		try {
			this.structuredProxyPushConsumer.offer_change(
					new EventType[] { new EventType("NewsBC", "Sports"),
							  new EventType("NewsBC", "Economy"),
							  new EventType("NewsBC", "Entertainment")},
					new EventType[0]);
			
			while(!Thread.interrupted()) {
				EventType eventType = new EventType();
				eventType.domain_name = "NewsBC";
				eventType.type_name = (i%3==0)?"Sports":((i%3==1)?"Economy":"Entertainment");
							
				EventHeader eventHeader = new EventHeader();
				
				eventHeader.fixed_header = new FixedEventHeader();
				eventHeader.fixed_header.event_type = eventType;
				eventHeader.fixed_header.event_name = "Event No. " + i++;
				
				eventHeader.variable_header = new Property[1];
				eventHeader.variable_header[0] = new Property();
				eventHeader.variable_header[0].name = Priority.value;
				eventHeader.variable_header[0].value = orb.create_any();
				short priority = (short)(DefaultPriority.value + new Random().nextInt(3));
				eventHeader.variable_header[0].value.insert_short(priority);
				
				StructuredEvent event = new StructuredEvent();
				event.header = eventHeader;
				event.filterable_data = new Property[0];
				event.remainder_of_body = orb.create_any();
				event.remainder_of_body.insert_string(
						eventType.type_name + " News, Priority " + priority);
				
				this.structuredProxyPushConsumer.push_structured_event(event);
				Thread.sleep(3000);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (Disconnected e) {
			e.printStackTrace();
		} catch (InvalidEventType e) {
			e.printStackTrace();
		}
		
		this.structuredProxyPushConsumer.disconnect_structured_push_consumer();
		this.orb.shutdown(true);
	}
}

Client代码
1）StructuredPushConsumerImpl.java

package learning.corba.notification.newsbroadcast.client;

import org.omg.CORBA.TCKind;
import org.omg.CosEventComm.Disconnected;
import org.omg.CosNotification.EventType;
import org.omg.CosNotification.Property;
import org.omg.CosNotification.StructuredEvent;
import org.omg.CosNotifyComm.InvalidEventType;
import org.omg.CosNotifyComm.StructuredPushConsumerPOA;

public class StructuredPushConsumerImpl extends StructuredPushConsumerPOA {

	private String identifier;
	
	public StructuredPushConsumerImpl(String identifier) {
		this.identifier = identifier;
	}

	public void disconnect_structured_push_consumer() {
		System.out.println("disconnect_structured_push_consumer");
		
	}

	public void push_structured_event(StructuredEvent event) throws Disconnected {
		StringBuilder sb = new StringBuilder();
		sb.append("[");
		sb.append(identifier);
		sb.append("]");
		sb.append(event.header.fixed_header.event_type.domain_name);
		sb.append(" ");
		sb.append(event.header.fixed_header.event_type.type_name);
		sb.append(",");
		sb.append(event.header.fixed_header.event_name);
		sb.append(",");
		for (Property prop : event.header.variable_header) {
			sb.append(prop.name);
			sb.append("=");
			if (prop.value.type().kind().value() == TCKind._tk_short) {
				sb.append(prop.value.extract_short());
			}
			sb.append(",");
		}
		
		if (event.remainder_of_body.type().kind().value() == TCKind._tk_string) {
			sb.append(event.remainder_of_body.extract_string());
		}
		System.out.println(sb.toString());		
	}

	public void offer_change(EventType[] added, EventType[] removed)
			throws InvalidEventType {
		throw new UnsupportedOperationException();		
	}

}

2）NewsBCNotifClient.java

package learning.corba.notification.newsbroadcast.client;

import java.util.Properties;

import org.omg.CORBA.IntHolder;
import org.omg.CORBA.ORB;
import org.omg.CORBA.UserException;
import org.omg.CosNotification.EventType;
import org.omg.CosNotifyChannelAdmin.ClientType;
import org.omg.CosNotifyChannelAdmin.ConsumerAdmin;
import org.omg.CosNotifyChannelAdmin.EventChannel;
import org.omg.CosNotifyChannelAdmin.EventChannelHelper;
import org.omg.CosNotifyChannelAdmin.InterFilterGroupOperator;
import org.omg.CosNotifyChannelAdmin.ObtainInfoMode;
import org.omg.CosNotifyChannelAdmin.StructuredProxyPushSupplierHelper;
import org.omg.CosNaming.NamingContextExt;
import org.omg.CosNaming.NamingContextExtHelper;
import org.omg.CosNotifyChannelAdmin.StructuredProxyPushSupplier;
import org.omg.CosNotifyComm.StructuredPushConsumerHelper;
import org.omg.CosNotifyFilter.ConstraintExp;
import org.omg.CosNotifyFilter.Filter;
import org.omg.CosNotifyFilter.FilterFactory;
import org.omg.PortableServer.POA;
import org.omg.PortableServer.POAHelper;

public class NewsBCNotifClient{

	public static void main(String[] args) {
		try {
			Properties props = new Properties();
			props.put("org.omg.PortableInterceptor.ORBInitializerClass.bidir_init",
			  	 	  "org.jacorb.orb.giop.BiDirConnectionInitializer");
			
			ORB orb = ORB.init(args, props);
			POA rootPoa = POAHelper.narrow(
					orb.resolve_initial_references("RootPOA"));
			rootPoa.the_POAManager().activate();
			
			NamingContextExt ncExt = NamingContextExtHelper.narrow(
					orb.resolve_initial_references("NameService"));
			
			EventChannel eventChannel = EventChannelHelper.narrow(
					ncExt.resolve_str("NewsBC.NotifChannel"));
			
			createConsumerAllEvents(eventChannel, rootPoa);
			createConsumerSportsPriority1Events(eventChannel, rootPoa);
			
			orb.run();			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void createConsumerAllEvents(EventChannel eventChannel, POA rootPoa)
	throws UserException {
		StructuredPushConsumerImpl pushConsumerImpl = new StructuredPushConsumerImpl(
				"CosumerAllEvents");
		ConsumerAdmin consumerAdmin = eventChannel.default_consumer_admin();
		StructuredProxyPushSupplier proxyPushSupplier =
				StructuredProxyPushSupplierHelper.narrow(
						consumerAdmin.obtain_notification_push_supplier(
								ClientType.STRUCTURED_EVENT, new IntHolder()));
		
		FilterFactory filterFactory = eventChannel.default_filter_factory();
		
		//Within a filter, individual constraints are combined using OR semantics.
		//When multiple filters apply, OR is applied to the combination of results of filters.
		ConstraintExp constraint1 = new ConstraintExp(
				new EventType[]{new EventType("NewsBC", "Sports")}, "TRUE");
		ConstraintExp constraint2 = new ConstraintExp(
				new EventType[]{new EventType("NewsBC", "Economy")}, "TRUE");
		Filter filter1 = filterFactory.create_filter("EXTENDED_TCL");
        	filter1.add_constraints(new ConstraintExp[]{constraint1, constraint2});
        
		ConstraintExp constraint3 = new ConstraintExp(
				new EventType[]{new EventType("NewsBC", "Entertainment")}, "TRUE");
		Filter filter2 = filterFactory.create_filter("EXTENDED_TCL");
        	filter2.add_constraints(new ConstraintExp[]{constraint3});
        
        	proxyPushSupplier.add_filter(filter1);
        	proxyPushSupplier.add_filter(filter2);

		proxyPushSupplier.connect_structured_push_consumer(
				StructuredPushConsumerHelper.narrow(
						rootPoa.servant_to_reference(pushConsumerImpl)));
	}
	
	public static void createConsumerSportsPriority1Events(EventChannel eventChannel, POA rootPoa)
	throws UserException {
		StructuredPushConsumerImpl pushConsumerImpl = new StructuredPushConsumerImpl(
				"CosumerSportsPrio1");
		//When the results of filters associated with a proxy
		//and those of filters associated with its admin object are combined,
		//the actual Boolean operator depends on
		//the value of the MyOperator attribute in the admin object.
		ConsumerAdmin consumerAdmin = eventChannel.new_for_consumers(
				InterFilterGroupOperator.AND_OP, new IntHolder());
		System.out.println(consumerAdmin.MyOperator());
		StructuredProxyPushSupplier proxyPushSupplier =
				StructuredProxyPushSupplierHelper.narrow(
						consumerAdmin.obtain_notification_push_supplier(
								ClientType.STRUCTURED_EVENT, new IntHolder()));
		
		FilterFactory filterFactory = eventChannel.default_filter_factory();
		
		ConstraintExp constraint1 = new ConstraintExp(
				new EventType[]{new EventType("NewsBC", "Sports")},
				"$Priority<2");
		Filter filter1 = filterFactory.create_filter("EXTENDED_TCL");
        	filter1.add_constraints(new ConstraintExp[]{constraint1});
        	consumerAdmin.add_filter(filter1);
        
		ConstraintExp constraint2 = new ConstraintExp(
				new EventType[]{new EventType("NewsBC", "Sports")},
				"$.header.variable_header(Priority)>0");
		Filter filter2 = filterFactory.create_filter("EXTENDED_TCL");
        	filter2.add_constraints(new ConstraintExp[]{constraint2});
        	proxyPushSupplier.add_filter(filter2);
        
		proxyPushSupplier.connect_structured_push_consumer(
				StructuredPushConsumerHelper.narrow(
						rootPoa.servant_to_reference(pushConsumerImpl)));
		
		EventType[] eventTypes = proxyPushSupplier.obtain_offered_types(
				ObtainInfoMode.ALL_NOW_UPDATES_OFF);
		for (EventType eventType : eventTypes) {
			System.out.println(eventType.domain_name + "." + eventType.type_name);
		}
	}

}