ChannelRef now supports three kinds of endpoints:
- CHANNEL (which is an identifyable abstract endpoint),
- MESSAGE_LISTENER (which is a concrete message listener implementation), and
- JACOB_OBJECT (which is an OO implementation.
diff --git a/src/main/java/org/apache/ode/jacob/ChannelRef.java b/src/main/java/org/apache/ode/jacob/ChannelRef.java
index d741b47..70b6a1f 100644
--- a/src/main/java/org/apache/ode/jacob/ChannelRef.java
+++ b/src/main/java/org/apache/ode/jacob/ChannelRef.java
@@ -29,7 +29,7 @@
*/
public class ChannelRef implements Serializable {
- public enum Type { JACOB_OBJECT, CHANNEL }
+ public enum Type { JACOB_OBJECT, CHANNEL, MESSAGE_LISTENER }
private static final long serialVersionUID = 1L;
@@ -37,7 +37,17 @@
private final Object target;
public ChannelRef(Object target) {
- type = target instanceof CommChannel ? Type.CHANNEL : Type.JACOB_OBJECT;
+ assert target != null;
+ if (target instanceof CommChannel) {
+ type = Type.CHANNEL;
+ } else if (target instanceof JacobObject) {
+ type = Type.JACOB_OBJECT;
+ } else if (target instanceof MessageListener) {
+ type = Type.MESSAGE_LISTENER;
+ } else {
+ throw new IllegalArgumentException("Unsupported endpoint reference");
+ }
+
this.target = target;
}
@@ -51,6 +61,8 @@
return (T)target;
} else if (type.equals(Type.CHANNEL) && CommChannel.class.isAssignableFrom(clazz)) {
return (T)target;
+ } else if (type.equals(Type.MESSAGE_LISTENER) && MessageListener.class.isAssignableFrom(clazz)) {
+ return (T)target;
}
return null;
diff --git a/src/main/java/org/apache/ode/jacob/soup/jackson/ChannelRefSerializer.java b/src/main/java/org/apache/ode/jacob/soup/jackson/ChannelRefSerializer.java
index c50bb00..4e226f1 100644
--- a/src/main/java/org/apache/ode/jacob/soup/jackson/ChannelRefSerializer.java
+++ b/src/main/java/org/apache/ode/jacob/soup/jackson/ChannelRefSerializer.java
@@ -19,6 +19,7 @@
package org.apache.ode.jacob.soup.jackson;
import java.io.IOException;
+import java.lang.reflect.Field;
import org.apache.ode.jacob.ChannelRef;
import org.apache.ode.jacob.JacobObject;
@@ -66,7 +67,13 @@
private void serializeContents(ChannelRef value, JsonGenerator jgen,
SerializerProvider provider) throws JsonGenerationException, IOException {
- jgen.writeObjectField("target", value.getEndpoint(value.getType() == ChannelRef.Type.CHANNEL
- ? CommChannel.class : JacobObject.class));
+ try {
+ Field targetField = ChannelRef.class.getDeclaredField("target");
+ targetField.setAccessible(true);
+ jgen.writeObjectField("target", targetField.get(value));
+
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
}
}
\ No newline at end of file
diff --git a/src/main/java/org/apache/ode/jacob/vpu/JacobVPU.java b/src/main/java/org/apache/ode/jacob/vpu/JacobVPU.java
index cf47558..b70e5ab 100644
--- a/src/main/java/org/apache/ode/jacob/vpu/JacobVPU.java
+++ b/src/main/java/org/apache/ode/jacob/vpu/JacobVPU.java
@@ -466,12 +466,23 @@
long ctime = System.currentTimeMillis();
try {
- JacobObject target = message.getTo().getEndpoint(JacobObject.class);
- if (target instanceof ReceiveProcess) {
- ((ReceiveProcess)target).onMessage(message);
- } else {
- ((Runnable)target).run();
- }
+ switch (message.getTo().getType()) {
+ case CHANNEL:
+ throw new UnsupportedOperationException();
+ case JACOB_OBJECT:
+ JacobObject target = message.getTo().getEndpoint(JacobObject.class);
+ if (target instanceof ReceiveProcess) {
+ ((ReceiveProcess)target).onMessage(message);
+ } else {
+ ((Runnable)target).run();
+ }
+ break;
+ case MESSAGE_LISTENER:
+ MessageListener ml = message.getTo().getEndpoint(MessageListener.class);
+ ml.onMessage(message);
+ break;
+ }
+
if (replyTo != null) {
sendMessage(ClassUtil.createMessage(replyTo, ClassUtil.SYNCH_RET_METHOD_ACTION, null, null));
}
