Problems implementing TypedPushConsumerPOA (org.omg.CosTypedEventComm)

Hi,

I'm testing the typed event channel of orbacus. I want to implement a typ=
ed push consumer and supplier. Therefore, I defined the IDL:

module model {
struct MyMessage{
	long id;
	string message;
};

interface MyMessageConsumerInterf {
	void process(in MyMessage myMsg);
};
};

Then, I implemented a class called TypedPushConsumerImpl which extends or=
g.omg.CosTypedEventComm.TypedPushConsumerPOA. The TypedPushConsumerImpl c=
lass does the following:
- connecting to channel (getting proxy supplier for MyMessageConsumerInte=
f Interface, connecting to that proxy)
- creates a MyMessageConsumerImpl instance, which implements the MyMessag=
eConsumerInterf Interface
- delegates the TypedEvents to this instance via the method get_typed_con=
sumer() which returns this instance
- implements the missing methods from the interface TypedPushConsumer

When I run this example everything works fine, but:
If I send a typed event to that consumer the consumer disconnects from th=
e  channel without any exception ! The called method process(MyMessage my=
Msg) is never entered ! The typedPushSupplier exits normally. I tried thi=
s szenario using VisiBroker and it works fine. Where is the prob ??

Thanks a lot,

Tim

PS: I attach you the main classes. If that's not enough, I'll send you my=
 whole project. Just let me know.

That's the code of the TypedPushConsumerImpl:
package impl;

import util.*;

import org.omg.CORBA.Any;
import org.omg.CORBA.Object;
import org.omg.CORBA.Policy;
import org.omg.CORBA.ORBPackage.InvalidName;
import org.omg.CosEventComm.Disconnected;
import org.omg.CosTypedEventComm.TypedPushConsumerPOA;
import org.omg.CosTypedEventComm.TypedPushConsumerHelper;
import org.omg.CosTypedEventChannelAdmin.NoSuchImplementation;
import org.omg.CosTypedEventChannelAdmin.TypedEventChannel;
import org.omg.CosTypedEventChannelAdmin.TypedConsumerAdmin;
import org.omg.CosEventChannelAdmin.*;
import org.omg.PortableServer.*;
import org.omg.PortableServer.POAManagerPackage.AdapterInactive;
import org.omg.PortableServer.POAPackage.*;

// this is the real TypedPushConsumer
// it handles the "standard communication"
// the "typed events" will be handled by MyMessageConsumerImpl (delegated=
)
public class TypedPushConsumerImpl extends TypedPushConsumerPOA {
	private TypedEventChannel tec =3D null;
	ProxyPushSupplier pps =3D null;=20
	org.omg.CORBA.Object typedConsumerRef =3D null;	// this is the MyMessage=
ConsumerImpl which
													// implements MyMessageConsumerInterf from IDL
	MyMessageConsumerImpl typedConsumerServant =3D null;											=09

=09
	public TypedPushConsumerImpl(TypedEventChannel tec) {
		this.tec =3D tec;
		typedConsumerServant =3D new MyMessageConsumerImpl();
	=09
		// Create a persistent POA for the consumer (MyMessageConsumerImpl)
		POA rootPOA =3D null;
		try {
			rootPOA =3D POAHelper.narrow(ChannelUtil.getORB().resolve_initial_refe=
rences("RootPOA"));
		} catch (InvalidName e) {
			System.out.println("TypedPushConsumerImpl: unable to resolve RootPOA")=
;
			e.printStackTrace();
		}
	=09
		// activate typedConsumerServant and get reference ...
		try {
			rootPOA.activate_object(typedConsumerServant);
			typedConsumerRef =3D rootPOA.servant_to_reference(typedConsumerServant=
);
		} catch (ServantAlreadyActive e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (WrongPolicy e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (ServantNotActive e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	=09
		// activate TypedPushConsumer (this) ...
		try {
			rootPOA.activate_object(this);
		} catch (ServantAlreadyActive e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (WrongPolicy e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	=09
		// activate POA(s)...
		try {
			rootPOA.the_POAManager().activate();=09
		} catch (AdapterInactive e) {
			System.out.println("TypedPushConsumerImpl: unable to activate RootPOA_=
Manager");
			e.printStackTrace();
		}
	=09
		// get ProxyPushSupplier from typed event channel ...=20
		TypedConsumerAdmin tca =3D tec.for_consumers();
		try {
			pps =3D tca.obtain_typed_push_supplier(MyMessageConsumerImpl.MESSAGE_C=
ONSUMER_INTERFACE);
		} catch (NoSuchImplementation e) {
			System.out.println("TypedPushConsumerImpl: could not obtain typed push=
 supplier " +
					"for IDL Interface " + MyMessageConsumerImpl.MESSAGE_CONSUMER_INTERF=
ACE + ".\n "+
					"(" + e.toString() + ")");
			System.exit(1);
		}
	=09
		// connect to proxyPushSupplier
		try {
			pps.connect_push_consumer(TypedPushConsumerHelper.narrow(rootPOA.serva=
nt_to_reference(this)));
			//pps.connect_push_consumer(_this());
		} catch (AlreadyConnected e) {
			System.out.println("TypedPushConsumerImpl: Already connected to channe=
l.");
			System.exit (1);
		} catch (TypeError e) {
			System.out.println("TypedPushConsumerImpl: TypeError while connecting =
to channel.");
			e.printStackTrace();
			System.exit(1);
		} catch (ServantNotActive e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (WrongPolicy e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
=09
	public Object get_typed_consumer() {
		return typedConsumerRef;
	}

	public void push(Any arg0) throws Disconnected {
	}

	public void disconnect_push_consumer() {
	=09
		System.out.println("disconnecting proxies and deactivating POAs ...");
		pps.disconnect_push_supplier();
		ChannelUtil.orbShutdown();
		System.out.println("... disconnected and ORB shut down.");
		return;
	}
=09
=09
	//	 available if notificaiton service is used=20
	// (import org.omg.CosTypedNotifyChannelAdmin.TypedEventChannel ...,
	//  and org.omg.CosTypedNotifyComm.TypedPushConsumer)
	/*
	public void offer_change(EventType[] arg0, EventType[] arg1)
			throws InvalidEventType {
		// TODO Auto-generated method stub

	}=09
	*/
}

And that's the supplier:

package impl;

import org.omg.CORBA.Object;
import org.omg.CosEventChannelAdmin.AlreadyConnected;
import org.omg.CosEventComm.Disconnected;
import org.omg.CosEventComm.PushSupplierPOA;
import org.omg.CosTypedEventChannelAdmin.*;

import util.ChannelUtil;
import model.*;

public class MyMessageSupplierImpl extends PushSupplierPOA {
=09
	private TypedProxyPushConsumer proxyConsumer =3D null;
	private TypedEventChannel tec;
	private MyMessageConsumerInterf msgConsumer =3D null;
=09
	public MyMessageSupplierImpl(TypedEventChannel tec) {
		this.tec =3D tec;
		try {
			System.out.println("getting proxyConsumer ...");
		=09
			proxyConsumer =3D tec.for_suppliers().obtain_typed_push_consumer(MyMes=
sageConsumerImpl.MESSAGE_CONSUMER_INTERFACE);
			System.out.println("... SUCCESS");
		} catch (InterfaceNotSupported e) {
			System.out.println("MyMessageSupplierImpl: Exception while connecting =
to channel." +
					"("+ e.toString() + ")");
			System.exit(1);
		}
		try {
			System.out.println("connecting to proxyConsumer ...");
			// if messageConsumer should be able to be informed when the proxy dis=
connects,
			// a POA for this supplier has to be created and .this() should be pas=
sed to=20
			// the following method
			// (If the supplier does not need to be informed if its proxy disconne=
cts from the
			// channel, the supplier can connect a null to the typed proxy consume=
r.)
			proxyConsumer.connect_push_supplier(null);
			Object obj =3D proxyConsumer.get_typed_consumer();
			msgConsumer =3D MyMessageConsumerInterfHelper.narrow(obj);
			System.out.println("... SUCCESS");
		} catch (AlreadyConnected e) {
			System.out.println("MyMessageSupplierImpl: Already connected to channe=
l.");
			System.exit (1);
		}
	}
=09
	public void push(MyMessage myMsg) {
		msgConsumer.process(myMsg);
	}
=09
	public void disconnect_push_supplier() {
		System.out.println("disconnecting from channel...");
		if (msgConsumer !=3D null) {
			System.out.println("flushing events from msgConsumer ...");
			msgConsumer._non_existent();
		}

		if (proxyConsumer !=3D null) {
			System.out.println("disconnecting from proxyConsumer ...");
			proxyConsumer.disconnect_push_consumer();
		}
		ChannelUtil.orbShutdown();
		System.out.println("disconnected and ORB shut down.");
	}

}

finally, here's the MyMessageInterf implementation:
package impl;

import model.*;

import org.omg.CosEventComm.Disconnected;

public class MyMessageConsumerImpl extends MyMessageConsumerInterfPOA{
	public static final String MESSAGE_CONSUMER_INTERFACE =3D "IDL:model/MyM=
essageConsumerInterf:1.0";
=09
	public void process(MyMessage myMsg) {
		System.out.println("got event !");
		System.out.println("Received Message: ID =3D" + myMsg.id + "Message =3D=
 " + myMsg.message +".");
	}


}




--=20


Der GMX SmartSurfer hilft bis zu 70% Ihrer Onlinekosten zu sparen!
Ideal f=FCr Modem und ISDN: http://www.gmx.net/de/go/smartsurfer
_______________________________________________
OB-Users Mailing List - [email protected]
http://mail.ooc.nf.ca/mailman/listinfo/ob-users
Visit our support FAQ before you send a message.
http://www.orbacus.com/faq/support.html