{
+ private se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol result;
+
+ // Construct using se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.newBuilder()
+ private Builder() {}
+
+ private static Builder create() {
+ Builder builder = new Builder();
+ builder.result = new se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol();
+ return builder;
+ }
+
+ protected se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol internalGetResult() {
+ return result;
+ }
+
+ public Builder clear() {
+ if (result == null) {
+ throw new IllegalStateException(
+ "Cannot call clear() after build().");
+ }
+ result = new se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol();
+ return this;
+ }
+
+ public Builder clone() {
+ return create().mergeFrom(result);
+ }
+
+ public com.google.protobuf.Descriptors.Descriptor
+ getDescriptorForType() {
+ return se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.getDescriptor();
+ }
+
+ public se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol getDefaultInstanceForType() {
+ return se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.getDefaultInstance();
+ }
+
+ public boolean isInitialized() {
+ return result.isInitialized();
+ }
+ public se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol build() {
+ if (result != null && !isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return buildPartial();
+ }
+
+ private se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol buildParsed()
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ if (!isInitialized()) {
+ throw newUninitializedMessageException(
+ result).asInvalidProtocolBufferException();
+ }
+ return buildPartial();
+ }
+
+ public se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol buildPartial() {
+ if (result == null) {
+ throw new IllegalStateException(
+ "build() has already been called on this Builder.");
+ }
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol returnMe = result;
+ result = null;
+ return returnMe;
+ }
+
+ public Builder mergeFrom(com.google.protobuf.Message other) {
+ if (other instanceof se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol) {
+ return mergeFrom((se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol other) {
+ if (other == se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.getDefaultInstance()) return this;
+ if (other.hasActorRef()) {
+ mergeActorRef(other.getActorRef());
+ }
+ if (other.hasInterfaceName()) {
+ setInterfaceName(other.getInterfaceName());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ return this;
+ }
+
+ public Builder mergeFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ com.google.protobuf.UnknownFieldSet.Builder unknownFields =
+ com.google.protobuf.UnknownFieldSet.newBuilder(
+ this.getUnknownFields());
+ while (true) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ this.setUnknownFields(unknownFields.build());
+ return this;
+ default: {
+ if (!parseUnknownField(input, unknownFields,
+ extensionRegistry, tag)) {
+ this.setUnknownFields(unknownFields.build());
+ return this;
+ }
+ break;
+ }
+ case 10: {
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.Builder subBuilder = se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.newBuilder();
+ if (hasActorRef()) {
+ subBuilder.mergeFrom(getActorRef());
+ }
+ input.readMessage(subBuilder, extensionRegistry);
+ setActorRef(subBuilder.buildPartial());
+ break;
+ }
+ case 18: {
+ setInterfaceName(input.readString());
+ break;
+ }
+ }
+ }
+ }
+
+
+ // required .SerializedActorRefProtocol actorRef = 1;
+ public boolean hasActorRef() {
+ return result.hasActorRef();
+ }
+ public se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol getActorRef() {
+ return result.getActorRef();
+ }
+ public Builder setActorRef(se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ result.hasActorRef = true;
+ result.actorRef_ = value;
+ return this;
+ }
+ public Builder setActorRef(se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.Builder builderForValue) {
+ result.hasActorRef = true;
+ result.actorRef_ = builderForValue.build();
+ return this;
+ }
+ public Builder mergeActorRef(se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol value) {
+ if (result.hasActorRef() &&
+ result.actorRef_ != se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.getDefaultInstance()) {
+ result.actorRef_ =
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.newBuilder(result.actorRef_).mergeFrom(value).buildPartial();
+ } else {
+ result.actorRef_ = value;
+ }
+ result.hasActorRef = true;
+ return this;
+ }
+ public Builder clearActorRef() {
+ result.hasActorRef = false;
+ result.actorRef_ = se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.getDefaultInstance();
+ return this;
+ }
+
+ // required string interfaceName = 2;
+ public boolean hasInterfaceName() {
+ return result.hasInterfaceName();
+ }
+ public java.lang.String getInterfaceName() {
+ return result.getInterfaceName();
+ }
+ public Builder setInterfaceName(java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ result.hasInterfaceName = true;
+ result.interfaceName_ = value;
+ return this;
+ }
+ public Builder clearInterfaceName() {
+ result.hasInterfaceName = false;
+ result.interfaceName_ = getDefaultInstance().getInterfaceName();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:SerializedTypedActorRefProtocol)
+ }
+
+ static {
+ defaultInstance = new SerializedTypedActorRefProtocol(true);
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.internalForceInit();
+ defaultInstance.initFields();
+ }
+
+ // @@protoc_insertion_point(class_scope:SerializedTypedActorRefProtocol)
+ }
+
public static final class MessageProtocol extends
com.google.protobuf.GeneratedMessage {
// Use MessageProtocol.newBuilder() to construct.
@@ -5848,11 +6556,21 @@ public final class RemoteProtocol {
private static
com.google.protobuf.GeneratedMessage.FieldAccessorTable
internal_static_RemoteActorRefProtocol_fieldAccessorTable;
+ private static com.google.protobuf.Descriptors.Descriptor
+ internal_static_RemoteTypedActorRefProtocol_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_RemoteTypedActorRefProtocol_fieldAccessorTable;
private static com.google.protobuf.Descriptors.Descriptor
internal_static_SerializedActorRefProtocol_descriptor;
private static
com.google.protobuf.GeneratedMessage.FieldAccessorTable
internal_static_SerializedActorRefProtocol_fieldAccessorTable;
+ private static com.google.protobuf.Descriptors.Descriptor
+ internal_static_SerializedTypedActorRefProtocol_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_SerializedTypedActorRefProtocol_fieldAccessorTable;
private static com.google.protobuf.Descriptors.Descriptor
internal_static_MessageProtocol_descriptor;
private static
@@ -5915,52 +6633,57 @@ public final class RemoteProtocol {
"\n\024RemoteProtocol.proto\"\204\001\n\026RemoteActorRe" +
"fProtocol\022\032\n\022classOrServiceName\030\001 \002(\t\022\026\n" +
"\016actorClassname\030\002 \002(\t\022%\n\013homeAddress\030\003 \002" +
- "(\0132\020.AddressProtocol\022\017\n\007timeout\030\004 \001(\004\"\217\003" +
- "\n\032SerializedActorRefProtocol\022\033\n\004uuid\030\001 \002" +
- "(\0132\r.UuidProtocol\022\n\n\002id\030\002 \002(\t\022\026\n\016actorCl" +
- "assname\030\003 \002(\t\022)\n\017originalAddress\030\004 \002(\0132\020" +
- ".AddressProtocol\022\025\n\ractorInstance\030\005 \001(\014\022" +
- "\033\n\023serializerClassname\030\006 \001(\t\022\024\n\014isTransa" +
- "ctor\030\007 \001(\010\022\017\n\007timeout\030\010 \001(\004\022\026\n\016receiveTi",
- "meout\030\t \001(\004\022%\n\tlifeCycle\030\n \001(\0132\022.LifeCyc" +
- "leProtocol\022+\n\nsupervisor\030\013 \001(\0132\027.RemoteA" +
- "ctorRefProtocol\022\024\n\014hotswapStack\030\014 \001(\014\022(\n" +
- "\010messages\030\r \003(\0132\026.RemoteRequestProtocol\"" +
- "r\n\017MessageProtocol\0225\n\023serializationSchem" +
- "e\030\001 \002(\0162\030.SerializationSchemeType\022\017\n\007mes" +
- "sage\030\002 \002(\014\022\027\n\017messageManifest\030\003 \001(\014\"\255\001\n\021" +
- "ActorInfoProtocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidPr" +
- "otocol\022\016\n\006target\030\002 \002(\t\022\017\n\007timeout\030\003 \002(\004\022" +
- "\035\n\tactorType\030\004 \002(\0162\n.ActorType\022/\n\016typedA",
- "ctorInfo\030\005 \001(\0132\027.TypedActorInfoProtocol\022" +
- "\n\n\002id\030\006 \001(\t\";\n\026TypedActorInfoProtocol\022\021\n" +
- "\tinterface\030\001 \002(\t\022\016\n\006method\030\002 \002(\t\"\212\002\n\025Rem" +
- "oteRequestProtocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidP" +
- "rotocol\022!\n\007message\030\002 \002(\0132\020.MessageProtoc" +
- "ol\022%\n\tactorInfo\030\003 \002(\0132\022.ActorInfoProtoco" +
- "l\022\020\n\010isOneWay\030\004 \002(\010\022%\n\016supervisorUuid\030\005 " +
- "\001(\0132\r.UuidProtocol\022\'\n\006sender\030\006 \001(\0132\027.Rem" +
- "oteActorRefProtocol\022(\n\010metadata\030\007 \003(\0132\026." +
- "MetadataEntryProtocol\"\364\001\n\023RemoteReplyPro",
- "tocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidProtocol\022!\n\007me" +
- "ssage\030\002 \001(\0132\020.MessageProtocol\022%\n\texcepti" +
- "on\030\003 \001(\0132\022.ExceptionProtocol\022%\n\016supervis" +
- "orUuid\030\004 \001(\0132\r.UuidProtocol\022\017\n\007isActor\030\005" +
- " \002(\010\022\024\n\014isSuccessful\030\006 \002(\010\022(\n\010metadata\030\007" +
- " \003(\0132\026.MetadataEntryProtocol\")\n\014UuidProt" +
- "ocol\022\014\n\004high\030\001 \002(\004\022\013\n\003low\030\002 \002(\004\"3\n\025Metad" +
- "ataEntryProtocol\022\013\n\003key\030\001 \002(\t\022\r\n\005value\030\002" +
- " \002(\014\"6\n\021LifeCycleProtocol\022!\n\tlifeCycle\030\001" +
- " \002(\0162\016.LifeCycleType\"1\n\017AddressProtocol\022",
- "\020\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"7\n\021Excep" +
- "tionProtocol\022\021\n\tclassname\030\001 \002(\t\022\017\n\007messa" +
- "ge\030\002 \002(\t*=\n\tActorType\022\017\n\013SCALA_ACTOR\020\001\022\016" +
- "\n\nJAVA_ACTOR\020\002\022\017\n\013TYPED_ACTOR\020\003*]\n\027Seria" +
- "lizationSchemeType\022\010\n\004JAVA\020\001\022\013\n\007SBINARY\020" +
- "\002\022\016\n\nSCALA_JSON\020\003\022\r\n\tJAVA_JSON\020\004\022\014\n\010PROT" +
- "OBUF\020\005*-\n\rLifeCycleType\022\r\n\tPERMANENT\020\001\022\r" +
- "\n\tTEMPORARY\020\002B-\n)se.scalablesolutions.ak" +
- "ka.remote.protocolH\001"
+ "(\0132\020.AddressProtocol\022\017\n\007timeout\030\004 \001(\004\"_\n" +
+ "\033RemoteTypedActorRefProtocol\022)\n\010actorRef" +
+ "\030\001 \002(\0132\027.RemoteActorRefProtocol\022\025\n\rinter" +
+ "faceName\030\002 \002(\t\"\217\003\n\032SerializedActorRefPro" +
+ "tocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidProtocol\022\n\n\002id" +
+ "\030\002 \002(\t\022\026\n\016actorClassname\030\003 \002(\t\022)\n\017origin" +
+ "alAddress\030\004 \002(\0132\020.AddressProtocol\022\025\n\ract",
+ "orInstance\030\005 \001(\014\022\033\n\023serializerClassname\030" +
+ "\006 \001(\t\022\024\n\014isTransactor\030\007 \001(\010\022\017\n\007timeout\030\010" +
+ " \001(\004\022\026\n\016receiveTimeout\030\t \001(\004\022%\n\tlifeCycl" +
+ "e\030\n \001(\0132\022.LifeCycleProtocol\022+\n\nsuperviso" +
+ "r\030\013 \001(\0132\027.RemoteActorRefProtocol\022\024\n\014hots" +
+ "wapStack\030\014 \001(\014\022(\n\010messages\030\r \003(\0132\026.Remot" +
+ "eRequestProtocol\"g\n\037SerializedTypedActor" +
+ "RefProtocol\022-\n\010actorRef\030\001 \002(\0132\033.Serializ" +
+ "edActorRefProtocol\022\025\n\rinterfaceName\030\002 \002(" +
+ "\t\"r\n\017MessageProtocol\0225\n\023serializationSch",
+ "eme\030\001 \002(\0162\030.SerializationSchemeType\022\017\n\007m" +
+ "essage\030\002 \002(\014\022\027\n\017messageManifest\030\003 \001(\014\"\255\001" +
+ "\n\021ActorInfoProtocol\022\033\n\004uuid\030\001 \002(\0132\r.Uuid" +
+ "Protocol\022\016\n\006target\030\002 \002(\t\022\017\n\007timeout\030\003 \002(" +
+ "\004\022\035\n\tactorType\030\004 \002(\0162\n.ActorType\022/\n\016type" +
+ "dActorInfo\030\005 \001(\0132\027.TypedActorInfoProtoco" +
+ "l\022\n\n\002id\030\006 \001(\t\";\n\026TypedActorInfoProtocol\022" +
+ "\021\n\tinterface\030\001 \002(\t\022\016\n\006method\030\002 \002(\t\"\212\002\n\025R" +
+ "emoteRequestProtocol\022\033\n\004uuid\030\001 \002(\0132\r.Uui" +
+ "dProtocol\022!\n\007message\030\002 \002(\0132\020.MessageProt",
+ "ocol\022%\n\tactorInfo\030\003 \002(\0132\022.ActorInfoProto" +
+ "col\022\020\n\010isOneWay\030\004 \002(\010\022%\n\016supervisorUuid\030" +
+ "\005 \001(\0132\r.UuidProtocol\022\'\n\006sender\030\006 \001(\0132\027.R" +
+ "emoteActorRefProtocol\022(\n\010metadata\030\007 \003(\0132" +
+ "\026.MetadataEntryProtocol\"\364\001\n\023RemoteReplyP" +
+ "rotocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidProtocol\022!\n\007" +
+ "message\030\002 \001(\0132\020.MessageProtocol\022%\n\texcep" +
+ "tion\030\003 \001(\0132\022.ExceptionProtocol\022%\n\016superv" +
+ "isorUuid\030\004 \001(\0132\r.UuidProtocol\022\017\n\007isActor" +
+ "\030\005 \002(\010\022\024\n\014isSuccessful\030\006 \002(\010\022(\n\010metadata",
+ "\030\007 \003(\0132\026.MetadataEntryProtocol\")\n\014UuidPr" +
+ "otocol\022\014\n\004high\030\001 \002(\004\022\013\n\003low\030\002 \002(\004\"3\n\025Met" +
+ "adataEntryProtocol\022\013\n\003key\030\001 \002(\t\022\r\n\005value" +
+ "\030\002 \002(\014\"6\n\021LifeCycleProtocol\022!\n\tlifeCycle" +
+ "\030\001 \002(\0162\016.LifeCycleType\"1\n\017AddressProtoco" +
+ "l\022\020\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"7\n\021Exc" +
+ "eptionProtocol\022\021\n\tclassname\030\001 \002(\t\022\017\n\007mes" +
+ "sage\030\002 \002(\t*=\n\tActorType\022\017\n\013SCALA_ACTOR\020\001" +
+ "\022\016\n\nJAVA_ACTOR\020\002\022\017\n\013TYPED_ACTOR\020\003*]\n\027Ser" +
+ "ializationSchemeType\022\010\n\004JAVA\020\001\022\013\n\007SBINAR",
+ "Y\020\002\022\016\n\nSCALA_JSON\020\003\022\r\n\tJAVA_JSON\020\004\022\014\n\010PR" +
+ "OTOBUF\020\005*-\n\rLifeCycleType\022\r\n\tPERMANENT\020\001" +
+ "\022\r\n\tTEMPORARY\020\002B-\n)se.scalablesolutions." +
+ "akka.remote.protocolH\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -5975,16 +6698,32 @@ public final class RemoteProtocol {
new java.lang.String[] { "ClassOrServiceName", "ActorClassname", "HomeAddress", "Timeout", },
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol.Builder.class);
- internal_static_SerializedActorRefProtocol_descriptor =
+ internal_static_RemoteTypedActorRefProtocol_descriptor =
getDescriptor().getMessageTypes().get(1);
+ internal_static_RemoteTypedActorRefProtocol_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_RemoteTypedActorRefProtocol_descriptor,
+ new java.lang.String[] { "ActorRef", "InterfaceName", },
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteTypedActorRefProtocol.class,
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteTypedActorRefProtocol.Builder.class);
+ internal_static_SerializedActorRefProtocol_descriptor =
+ getDescriptor().getMessageTypes().get(2);
internal_static_SerializedActorRefProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_SerializedActorRefProtocol_descriptor,
new java.lang.String[] { "Uuid", "Id", "ActorClassname", "OriginalAddress", "ActorInstance", "SerializerClassname", "IsTransactor", "Timeout", "ReceiveTimeout", "LifeCycle", "Supervisor", "HotswapStack", "Messages", },
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.Builder.class);
+ internal_static_SerializedTypedActorRefProtocol_descriptor =
+ getDescriptor().getMessageTypes().get(3);
+ internal_static_SerializedTypedActorRefProtocol_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_SerializedTypedActorRefProtocol_descriptor,
+ new java.lang.String[] { "ActorRef", "InterfaceName", },
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.class,
+ se.scalablesolutions.akka.remote.protocol.RemoteProtocol.SerializedTypedActorRefProtocol.Builder.class);
internal_static_MessageProtocol_descriptor =
- getDescriptor().getMessageTypes().get(2);
+ getDescriptor().getMessageTypes().get(4);
internal_static_MessageProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_MessageProtocol_descriptor,
@@ -5992,7 +6731,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.MessageProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.MessageProtocol.Builder.class);
internal_static_ActorInfoProtocol_descriptor =
- getDescriptor().getMessageTypes().get(3);
+ getDescriptor().getMessageTypes().get(5);
internal_static_ActorInfoProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_ActorInfoProtocol_descriptor,
@@ -6000,7 +6739,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.ActorInfoProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.ActorInfoProtocol.Builder.class);
internal_static_TypedActorInfoProtocol_descriptor =
- getDescriptor().getMessageTypes().get(4);
+ getDescriptor().getMessageTypes().get(6);
internal_static_TypedActorInfoProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_TypedActorInfoProtocol_descriptor,
@@ -6008,7 +6747,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.TypedActorInfoProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.TypedActorInfoProtocol.Builder.class);
internal_static_RemoteRequestProtocol_descriptor =
- getDescriptor().getMessageTypes().get(5);
+ getDescriptor().getMessageTypes().get(7);
internal_static_RemoteRequestProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_RemoteRequestProtocol_descriptor,
@@ -6016,7 +6755,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteRequestProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteRequestProtocol.Builder.class);
internal_static_RemoteReplyProtocol_descriptor =
- getDescriptor().getMessageTypes().get(6);
+ getDescriptor().getMessageTypes().get(8);
internal_static_RemoteReplyProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_RemoteReplyProtocol_descriptor,
@@ -6024,7 +6763,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteReplyProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.RemoteReplyProtocol.Builder.class);
internal_static_UuidProtocol_descriptor =
- getDescriptor().getMessageTypes().get(7);
+ getDescriptor().getMessageTypes().get(9);
internal_static_UuidProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_UuidProtocol_descriptor,
@@ -6032,7 +6771,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.UuidProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.UuidProtocol.Builder.class);
internal_static_MetadataEntryProtocol_descriptor =
- getDescriptor().getMessageTypes().get(8);
+ getDescriptor().getMessageTypes().get(10);
internal_static_MetadataEntryProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_MetadataEntryProtocol_descriptor,
@@ -6040,7 +6779,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.MetadataEntryProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.MetadataEntryProtocol.Builder.class);
internal_static_LifeCycleProtocol_descriptor =
- getDescriptor().getMessageTypes().get(9);
+ getDescriptor().getMessageTypes().get(11);
internal_static_LifeCycleProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_LifeCycleProtocol_descriptor,
@@ -6048,7 +6787,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.LifeCycleProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.LifeCycleProtocol.Builder.class);
internal_static_AddressProtocol_descriptor =
- getDescriptor().getMessageTypes().get(10);
+ getDescriptor().getMessageTypes().get(12);
internal_static_AddressProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_AddressProtocol_descriptor,
@@ -6056,7 +6795,7 @@ public final class RemoteProtocol {
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.AddressProtocol.class,
se.scalablesolutions.akka.remote.protocol.RemoteProtocol.AddressProtocol.Builder.class);
internal_static_ExceptionProtocol_descriptor =
- getDescriptor().getMessageTypes().get(11);
+ getDescriptor().getMessageTypes().get(13);
internal_static_ExceptionProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_ExceptionProtocol_descriptor,
diff --git a/akka-remote/src/main/protocol/RemoteProtocol.proto b/akka-remote/src/main/protocol/RemoteProtocol.proto
index 55c3ae35fd..1ea9d8f986 100644
--- a/akka-remote/src/main/protocol/RemoteProtocol.proto
+++ b/akka-remote/src/main/protocol/RemoteProtocol.proto
@@ -22,6 +22,15 @@ message RemoteActorRefProtocol {
optional uint64 timeout = 4;
}
+/**
+ * Defines a remote ActorRef that "remembers" and uses its original typed Actor instance
+ * on the original node.
+ */
+message RemoteTypedActorRefProtocol {
+ required RemoteActorRefProtocol actorRef = 1;
+ required string interfaceName = 2;
+}
+
/**
* Defines a fully serialized remote ActorRef (with serialized Actor instance)
* that is about to be instantiated on the remote node. It is fully disconnected
@@ -43,6 +52,16 @@ message SerializedActorRefProtocol {
repeated RemoteRequestProtocol messages = 13;
}
+/**
+ * Defines a fully serialized remote ActorRef (with serialized typed actor instance)
+ * that is about to be instantiated on the remote node. It is fully disconnected
+ * from its original host.
+ */
+message SerializedTypedActorRefProtocol {
+ required SerializedActorRefProtocol actorRef = 1;
+ required string interfaceName = 2;
+}
+
/**
* Defines a message.
*/
diff --git a/akka-remote/src/main/scala/remote/RemoteServer.scala b/akka-remote/src/main/scala/remote/RemoteServer.scala
index bacaf22546..8784a8c81f 100644
--- a/akka-remote/src/main/scala/remote/RemoteServer.scala
+++ b/akka-remote/src/main/scala/remote/RemoteServer.scala
@@ -10,8 +10,7 @@ import java.util.concurrent.{ConcurrentHashMap, Executors}
import java.util.{Map => JMap}
import se.scalablesolutions.akka.actor.{
- Actor, TypedActor, ActorRef, IllegalActorStateException, RemoteActorSystemMessage}
-import se.scalablesolutions.akka.actor.{Uuid,uuidFrom}
+ Actor, TypedActor, ActorRef, IllegalActorStateException, RemoteActorSystemMessage,uuidFrom,Uuid}
import se.scalablesolutions.akka.actor.Actor._
import se.scalablesolutions.akka.util._
import se.scalablesolutions.akka.remote.protocol.RemoteProtocol._
@@ -68,6 +67,7 @@ object RemoteNode extends RemoteServer
* @author Jonas Bonér
*/
object RemoteServer {
+ val UUID_PREFIX = "uuid:"
val HOSTNAME = config.getString("akka.remote.server.hostname", "localhost")
val PORT = config.getInt("akka.remote.server.port", 9999)
@@ -124,19 +124,21 @@ object RemoteServer {
private class RemoteActorSet {
private[RemoteServer] val actors = new ConcurrentHashMap[String, ActorRef]
+ private[RemoteServer] val actorsByUuid = new ConcurrentHashMap[String, ActorRef]
private[RemoteServer] val typedActors = new ConcurrentHashMap[String, AnyRef]
+ private[RemoteServer] val typedActorsByUuid = new ConcurrentHashMap[String, AnyRef]
}
private val guard = new ReadWriteGuard
private val remoteActorSets = Map[Address, RemoteActorSet]()
private val remoteServers = Map[Address, RemoteServer]()
- private[akka] def registerActor(address: InetSocketAddress, uuid: Uuid, actor: ActorRef) = guard.withWriteGuard {
- actorsFor(RemoteServer.Address(address.getHostName, address.getPort)).actors.put(uuid.toString, actor)
+ private[akka] def registerActorByUuid(address: InetSocketAddress, uuid: String, actor: ActorRef) = guard.withWriteGuard {
+ actorsFor(RemoteServer.Address(address.getHostName, address.getPort)).actorsByUuid.put(uuid, actor)
}
- private[akka] def registerTypedActor(address: InetSocketAddress, uuid: Uuid, typedActor: AnyRef) = guard.withWriteGuard {
- actorsFor(RemoteServer.Address(address.getHostName, address.getPort)).typedActors.put(uuid.toString, typedActor)
+ private[akka] def registerTypedActorByUuid(address: InetSocketAddress, uuid: String, typedActor: AnyRef) = guard.withWriteGuard {
+ actorsFor(RemoteServer.Address(address.getHostName, address.getPort)).typedActors.put(uuid, typedActor)
}
private[akka] def getOrCreateServer(address: InetSocketAddress): RemoteServer = guard.withWriteGuard {
@@ -193,6 +195,7 @@ case class RemoteServerClientDisconnected(@BeanProperty val server: RemoteServer
* @author Jonas Bonér
*/
class RemoteServer extends Logging with ListenerManagement {
+ import RemoteServer._
def name = "RemoteServer@" + hostname + ":" + port
private[akka] var address = RemoteServer.Address(RemoteServer.HOSTNAME,RemoteServer.PORT)
@@ -284,10 +287,11 @@ class RemoteServer extends Logging with ListenerManagement {
* @param typedActor typed actor to register
*/
def registerTypedActor(id: String, typedActor: AnyRef): Unit = synchronized {
- val typedActors = RemoteServer.actorsFor(RemoteServer.Address(hostname, port)).typedActors
- if (!typedActors.contains(id)) {
- log.debug("Registering server side remote actor [%s] with id [%s] on [%s:%d]", typedActor.getClass.getName, id, hostname, port)
- typedActors.put(id, typedActor)
+ log.debug("Registering server side remote typed actor [%s] with id [%s]", typedActor.getClass.getName, id)
+ if (id.startsWith(UUID_PREFIX)) {
+ registerTypedActor(id.substring(UUID_PREFIX.length), typedActor, typedActorsByUuid())
+ } else {
+ registerTypedActor(id, typedActor, typedActors())
}
}
@@ -302,12 +306,27 @@ class RemoteServer extends Logging with ListenerManagement {
* NOTE: If you use this method to register your remote actor then you must unregister the actor by this ID yourself.
*/
def register(id: String, actorRef: ActorRef): Unit = synchronized {
+ log.debug("Registering server side remote actor [%s] with id [%s]", actorRef.actorClass.getName, id)
+ if (id.startsWith(UUID_PREFIX)) {
+ register(id.substring(UUID_PREFIX.length), actorRef, actorsByUuid())
+ } else {
+ register(id, actorRef, actors())
+ }
+ }
+
+ private def register(id: String, actorRef: ActorRef, registry: ConcurrentHashMap[String, ActorRef]) {
if (_isRunning) {
- val actorMap = actors()
- if (!actorMap.contains(id)) {
+ if (!registry.contains(id)) {
if (!actorRef.isRunning) actorRef.start
- log.debug("Registering server side remote actor [%s] with id [%s]", actorRef.actorClass.getName, id)
- actorMap.put(id, actorRef)
+ registry.put(id, actorRef)
+ }
+ }
+ }
+
+ private def registerTypedActor(id: String, typedActor: AnyRef, registry: ConcurrentHashMap[String, AnyRef]) {
+ if (_isRunning) {
+ if (!registry.contains(id)) {
+ registry.put(id, typedActor)
}
}
}
@@ -320,7 +339,7 @@ class RemoteServer extends Logging with ListenerManagement {
log.debug("Unregistering server side remote actor [%s] with id [%s:%s]", actorRef.actorClass.getName, actorRef.id, actorRef.uuid)
val actorMap = actors()
actorMap remove actorRef.id
- if (actorRef.registeredInRemoteNodeDuringSerialization) actorMap remove actorRef.uuid
+ if (actorRef.registeredInRemoteNodeDuringSerialization) actorsByUuid() remove actorRef.uuid
}
}
@@ -332,10 +351,15 @@ class RemoteServer extends Logging with ListenerManagement {
def unregister(id: String):Unit = synchronized {
if (_isRunning) {
log.info("Unregistering server side remote actor with id [%s]", id)
- val actorMap = actors()
- val actorRef = actorMap get id
- actorMap remove id
- if (actorRef.registeredInRemoteNodeDuringSerialization) actorMap remove actorRef.uuid
+ if (id.startsWith(UUID_PREFIX)) {
+ actorsByUuid().remove(id.substring(UUID_PREFIX.length))
+ } else {
+ val actorRef = actors().get(id)
+ if (actorRef.registeredInRemoteNodeDuringSerialization) {
+ actorsByUuid() remove actorRef.uuid
+ }
+ actors() remove id
+ }
}
}
@@ -347,8 +371,11 @@ class RemoteServer extends Logging with ListenerManagement {
def unregisterTypedActor(id: String):Unit = synchronized {
if (_isRunning) {
log.info("Unregistering server side remote typed actor with id [%s]", id)
- val registeredTypedActors = typedActors()
- registeredTypedActors.remove(id)
+ if (id.startsWith(UUID_PREFIX)) {
+ typedActorsByUuid().remove(id.substring(UUID_PREFIX.length))
+ } else {
+ typedActors().remove(id)
+ }
}
}
@@ -356,8 +383,10 @@ class RemoteServer extends Logging with ListenerManagement {
protected[akka] override def notifyListeners(message: => Any): Unit = super.notifyListeners(message)
- private[akka] def actors() = RemoteServer.actorsFor(address).actors
- private[akka] def typedActors() = RemoteServer.actorsFor(address).typedActors
+ private[akka] def actors() = RemoteServer.actorsFor(address).actors
+ private[akka] def actorsByUuid() = RemoteServer.actorsFor(address).actorsByUuid
+ private[akka] def typedActors() = RemoteServer.actorsFor(address).typedActors
+ private[akka] def typedActorsByUuid() = RemoteServer.actorsFor(address).typedActorsByUuid
}
object RemoteServerSslContext {
@@ -420,6 +449,7 @@ class RemoteServerHandler(
val openChannels: ChannelGroup,
val applicationLoader: Option[ClassLoader],
val server: RemoteServer) extends SimpleChannelUpstreamHandler with Logging {
+ import RemoteServer._
val AW_PROXY_PREFIX = "$$ProxiedByAW".intern
applicationLoader.foreach(MessageSerializer.setClassLoader(_))
@@ -478,11 +508,12 @@ class RemoteServerHandler(
private def handleRemoteRequestProtocol(request: RemoteRequestProtocol, channel: Channel) = {
log.debug("Received RemoteRequestProtocol[\n%s]", request.toString)
- val actorType = request.getActorInfo.getActorType
- if (actorType == SCALA_ACTOR) dispatchToActor(request, channel)
- else if (actorType == JAVA_ACTOR) throw new IllegalActorStateException("ActorType JAVA_ACTOR is currently not supported")
- else if (actorType == TYPED_ACTOR) dispatchToTypedActor(request, channel)
- else throw new IllegalActorStateException("Unknown ActorType [" + actorType + "]")
+ request.getActorInfo.getActorType match {
+ case SCALA_ACTOR => dispatchToActor(request, channel)
+ case TYPED_ACTOR => dispatchToTypedActor(request, channel)
+ case JAVA_ACTOR => throw new IllegalActorStateException("ActorType JAVA_ACTOR is currently not supported")
+ case other => throw new IllegalActorStateException("Unknown ActorType [" + other + "]")
+ }
}
private def dispatchToActor(request: RemoteRequestProtocol, channel: Channel) = {
@@ -505,10 +536,10 @@ class RemoteServerHandler(
override def onComplete(result: AnyRef) {
log.debug("Returning result from actor invocation [%s]", result)
val replyBuilder = RemoteReplyProtocol.newBuilder
- .setUuid(request.getUuid)
- .setMessage(MessageSerializer.serialize(result))
- .setIsSuccessful(true)
- .setIsActor(true)
+ .setUuid(request.getUuid)
+ .setMessage(MessageSerializer.serialize(result))
+ .setIsSuccessful(true)
+ .setIsActor(true)
if (request.hasSupervisorUuid) replyBuilder.setSupervisorUuid(request.getSupervisorUuid)
@@ -566,32 +597,23 @@ class RemoteServerHandler(
}
}
- /**
- * Find a registered actor by ID (default) or UUID.
- * Actors are registered by id apart from registering during serialization see SerializationProtocol.
- */
- private def findActorByIdOrUuid(id: String, uuid: Uuid) : ActorRef = {
- val registeredActors = server.actors()
- var actorRefOrNull = registeredActors get id
- if (actorRefOrNull eq null) {
- actorRefOrNull = registeredActors get uuid
- }
- actorRefOrNull
+ private def findActorById(id: String) : ActorRef = {
+ server.actors().get(id)
}
- /**
- * Find a registered typed actor by ID (default) or UUID.
- * Actors are registered by id apart from registering during serialization see SerializationProtocol.
- */
- private def findTypedActorByIdOrUUid(id: String, uuid: Uuid) : AnyRef = {
- val registeredActors = server.typedActors()
- var actorRefOrNull = registeredActors get id
- if (actorRefOrNull eq null) {
- actorRefOrNull = registeredActors get uuid
- }
- actorRefOrNull
+ private def findActorByUuid(uuid: String) : ActorRef = {
+ server.actorsByUuid().get(uuid)
}
+ private def findTypedActorById(id: String) : AnyRef = {
+ server.typedActors().get(id)
+ }
+
+ private def findTypedActorByUuid(uuid: String) : AnyRef = {
+ server.typedActorsByUuid().get(uuid)
+ }
+
+
/**
* Creates a new instance of the actor with name, uuid and timeout specified as arguments.
*
@@ -600,21 +622,25 @@ class RemoteServerHandler(
* Does not start the actor.
*/
private def createActor(actorInfo: ActorInfoProtocol): ActorRef = {
- val uuid = uuidFrom(actorInfo.getUuid.getHigh,actorInfo.getUuid.getLow)
+ val uuid = actorInfo.getUuid
val id = actorInfo.getId
val name = actorInfo.getTarget
val timeout = actorInfo.getTimeout
- val actorRefOrNull = findActorByIdOrUuid(id, uuid)
-
+ val actorRefOrNull = if (id.startsWith(UUID_PREFIX)) {
+ findActorByUuid(id.substring(UUID_PREFIX.length))
+ } else {
+ findActorById(id)
+ }
+
if (actorRefOrNull eq null) {
try {
log.info("Creating a new remote actor [%s:%s]", name, uuid)
val clazz = if (applicationLoader.isDefined) applicationLoader.get.loadClass(name)
else Class.forName(name)
val actorRef = Actor.actorOf(clazz.newInstance.asInstanceOf[Actor])
- actorRef.uuid = uuid
+ actorRef.uuid = uuidFrom(uuid.getHigh,uuid.getLow)
actorRef.id = id
actorRef.timeout = timeout
actorRef.remoteAddress = None
@@ -630,10 +656,14 @@ class RemoteServerHandler(
}
private def createTypedActor(actorInfo: ActorInfoProtocol): AnyRef = {
- val uuid = uuidFrom(actorInfo.getUuid.getHigh,actorInfo.getUuid.getLow)
+ val uuid = actorInfo.getUuid
val id = actorInfo.getId
- val typedActorOrNull = findTypedActorByIdOrUUid(id, uuid)
+ val typedActorOrNull = if (id.startsWith(UUID_PREFIX)) {
+ findTypedActorByUuid(id.substring(UUID_PREFIX.length))
+ } else {
+ findTypedActorById(id)
+ }
if (typedActorOrNull eq null) {
val typedActorInfo = actorInfo.getTypedActorInfo
diff --git a/akka-remote/src/main/scala/serialization/SerializationProtocol.scala b/akka-remote/src/main/scala/serialization/SerializationProtocol.scala
index 7997be128b..0e7f5ce732 100644
--- a/akka-remote/src/main/scala/serialization/SerializationProtocol.scala
+++ b/akka-remote/src/main/scala/serialization/SerializationProtocol.scala
@@ -4,11 +4,10 @@
package se.scalablesolutions.akka.serialization
-import se.scalablesolutions.akka.actor.{Actor, ActorRef, LocalActorRef, RemoteActorRef, IllegalActorStateException, ActorType}
import se.scalablesolutions.akka.stm.global._
import se.scalablesolutions.akka.stm.TransactionManagement._
import se.scalablesolutions.akka.stm.TransactionManagement
-import se.scalablesolutions.akka.remote.protocol.RemoteProtocol._
+import se.scalablesolutions.akka.dispatch.MessageInvocation
import se.scalablesolutions.akka.remote.{RemoteServer, MessageSerializer}
import se.scalablesolutions.akka.remote.protocol.RemoteProtocol.{ActorType => ActorTypeProtocol, _}
import ActorTypeProtocol._
@@ -16,6 +15,7 @@ import se.scalablesolutions.akka.config.{AllForOneStrategy, OneForOneStrategy, F
import se.scalablesolutions.akka.config.ScalaConfig._
import se.scalablesolutions.akka.actor.{uuidFrom,newUuid}
import com.google.protobuf.ByteString
+import se.scalablesolutions.akka.actor._
/**
* Type class definition for Actor Serialization
@@ -37,13 +37,14 @@ trait Format[T <: Actor] extends FromBinary[T] with ToBinary[T]
* Create a Format object with the client actor as the implementation of the type class
*
*
- * object BinaryFormatMyStatelessActor {
+ * object BinaryFormatMyStatelessActor {
* implicit object MyStatelessActorFormat extends StatelessActorFormat[MyStatelessActor]
* }
*
*/
trait StatelessActorFormat[T <: Actor] extends Format[T] {
def fromBinary(bytes: Array[Byte], act: T) = act
+
def toBinary(ac: T) = Array.empty[Byte]
}
@@ -54,16 +55,18 @@ trait StatelessActorFormat[T <: Actor] extends Format[T] {
* a serializer object
*
*
- * object BinaryFormatMyJavaSerializableActor {
- * implicit object MyJavaSerializableActorFormat extends SerializerBasedActorFormat[MyJavaSerializableActor] {
+ * object BinaryFormatMyJavaSerializableActor {
+ * implicit object MyJavaSerializableActorFormat extends SerializerBasedActorFormat[MyJavaSerializableActor] {
* val serializer = Serializer.Java
- * }
+ * }
* }
*
*/
trait SerializerBasedActorFormat[T <: Actor] extends Format[T] {
val serializer: Serializer
+
def fromBinary(bytes: Array[Byte], act: T) = serializer.fromBinary(bytes, Some(act.self.actorClass)).asInstanceOf[T]
+
def toBinary(ac: T) = serializer.toBinary(ac)
}
@@ -71,23 +74,22 @@ trait SerializerBasedActorFormat[T <: Actor] extends Format[T] {
* Module for local actor serialization.
*/
object ActorSerialization {
-
def fromBinary[T <: Actor](bytes: Array[Byte])(implicit format: Format[T]): ActorRef =
fromBinaryToLocalActorRef(bytes, format)
- def toBinary[T <: Actor](a: ActorRef)(implicit format: Format[T]): Array[Byte] =
- toSerializedActorRefProtocol(a, format).toByteArray
+ def toBinary[T <: Actor](a: ActorRef, serializeMailBox: Boolean = true)(implicit format: Format[T]): Array[Byte] =
+ toSerializedActorRefProtocol(a, format, serializeMailBox).toByteArray
// wrapper for implicits to be used by Java
def fromBinaryJ[T <: Actor](bytes: Array[Byte], format: Format[T]): ActorRef =
fromBinary(bytes)(format)
// wrapper for implicits to be used by Java
- def toBinaryJ[T <: Actor](a: ActorRef, format: Format[T]): Array[Byte] =
- toBinary(a)(format)
+ def toBinaryJ[T <: Actor](a: ActorRef, format: Format[T], srlMailBox: Boolean = true): Array[Byte] =
+ toBinary(a, srlMailBox)(format)
- private def toSerializedActorRefProtocol[T <: Actor](
- actorRef: ActorRef, format: Format[T]): SerializedActorRefProtocol = {
+ private[akka] def toSerializedActorRefProtocol[T <: Actor](
+ actorRef: ActorRef, format: Format[T], serializeMailBox: Boolean = true): SerializedActorRefProtocol = {
val lifeCycleProtocol: Option[LifeCycleProtocol] = {
def setScope(builder: LifeCycleProtocol.Builder, scope: Scope) = scope match {
case Permanent => builder.setLifeCycle(LifeCycleType.PERMANENT)
@@ -103,9 +105,9 @@ object ActorSerialization {
}
val originalAddress = AddressProtocol.newBuilder
- .setHostname(actorRef.homeAddress.getHostName)
- .setPort(actorRef.homeAddress.getPort)
- .build
+ .setHostname(actorRef.homeAddress.getHostName)
+ .setPort(actorRef.homeAddress.getPort)
+ .build
val builder = SerializedActorRefProtocol.newBuilder
.setUuid(UuidProtocol.newBuilder.setHigh(actorRef.uuid.getTime).setLow(actorRef.uuid.getClockSeqAndNode).build)
@@ -115,6 +117,30 @@ object ActorSerialization {
.setIsTransactor(actorRef.isTransactor)
.setTimeout(actorRef.timeout)
+
+ if (serializeMailBox == true) {
+ val messages =
+ actorRef.mailbox match {
+ case q: java.util.Queue[MessageInvocation] =>
+ val l = new scala.collection.mutable.ListBuffer[MessageInvocation]
+ val it = q.iterator
+ while (it.hasNext == true) l += it.next
+ l
+ }
+
+ val requestProtocols =
+ messages.map(m =>
+ RemoteActorSerialization.createRemoteRequestProtocolBuilder(
+ actorRef,
+ m.message,
+ false,
+ actorRef.getSender,
+ None,
+ ActorType.ScalaActor).build)
+
+ requestProtocols.foreach(rp => builder.addMessages(rp))
+ }
+
actorRef.receiveTimeout.foreach(builder.setReceiveTimeout(_))
builder.setActorInstance(ByteString.copyFrom(format.toBinary(actorRef.actor.asInstanceOf[T])))
lifeCycleProtocol.foreach(builder.setLifeCycle(_))
@@ -127,33 +153,33 @@ object ActorSerialization {
private def fromBinaryToLocalActorRef[T <: Actor](bytes: Array[Byte], format: Format[T]): ActorRef =
fromProtobufToLocalActorRef(SerializedActorRefProtocol.newBuilder.mergeFrom(bytes).build, format, None)
- private def fromProtobufToLocalActorRef[T <: Actor](
- protocol: SerializedActorRefProtocol, format: Format[T], loader: Option[ClassLoader]): ActorRef = {
+ private[akka] def fromProtobufToLocalActorRef[T <: Actor](
+ protocol: SerializedActorRefProtocol, format: Format[T], loader: Option[ClassLoader]): ActorRef = {
Actor.log.debug("Deserializing SerializedActorRefProtocol to LocalActorRef:\n" + protocol)
val serializer =
- if (format.isInstanceOf[SerializerBasedActorFormat[_]])
- Some(format.asInstanceOf[SerializerBasedActorFormat[_]].serializer)
- else None
+ if (format.isInstanceOf[SerializerBasedActorFormat[_]])
+ Some(format.asInstanceOf[SerializerBasedActorFormat[_]].serializer)
+ else None
val lifeCycle =
- if (protocol.hasLifeCycle) {
- val lifeCycleProtocol = protocol.getLifeCycle
- Some(if (lifeCycleProtocol.getLifeCycle == LifeCycleType.PERMANENT) LifeCycle(Permanent)
- else if (lifeCycleProtocol.getLifeCycle == LifeCycleType.TEMPORARY) LifeCycle(Temporary)
- else throw new IllegalActorStateException("LifeCycle type is not valid: " + lifeCycleProtocol.getLifeCycle))
- } else None
+ if (protocol.hasLifeCycle) {
+ val lifeCycleProtocol = protocol.getLifeCycle
+ Some(if (lifeCycleProtocol.getLifeCycle == LifeCycleType.PERMANENT) LifeCycle(Permanent)
+ else if (lifeCycleProtocol.getLifeCycle == LifeCycleType.TEMPORARY) LifeCycle(Temporary)
+ else throw new IllegalActorStateException("LifeCycle type is not valid: " + lifeCycleProtocol.getLifeCycle))
+ } else None
val supervisor =
- if (protocol.hasSupervisor)
- Some(RemoteActorSerialization.fromProtobufToRemoteActorRef(protocol.getSupervisor, loader))
- else None
+ if (protocol.hasSupervisor)
+ Some(RemoteActorSerialization.fromProtobufToRemoteActorRef(protocol.getSupervisor, loader))
+ else None
val hotswap =
- if (serializer.isDefined && protocol.hasHotswapStack) Some(serializer.get
+ if (serializer.isDefined && protocol.hasHotswapStack) Some(serializer.get
.fromBinary(protocol.getHotswapStack.toByteArray, Some(classOf[PartialFunction[Any, Unit]]))
.asInstanceOf[PartialFunction[Any, Unit]])
- else None
+ else None
val classLoader = loader.getOrElse(getClass.getClassLoader)
@@ -195,9 +221,9 @@ object RemoteActorSerialization {
def fromBinaryToRemoteActorRef(bytes: Array[Byte]): ActorRef =
fromProtobufToRemoteActorRef(RemoteActorRefProtocol.newBuilder.mergeFrom(bytes).build, None)
- /**
- * Deserializes a byte array (Array[Byte]) into an RemoteActorRef instance.
- */
+ /**
+ * Deserializes a byte array (Array[Byte]) into an RemoteActorRef instance.
+ */
def fromBinaryToRemoteActorRef(bytes: Array[Byte], loader: ClassLoader): ActorRef =
fromProtobufToRemoteActorRef(RemoteActorRefProtocol.newBuilder.mergeFrom(bytes).build, Some(loader))
@@ -226,7 +252,7 @@ object RemoteActorSerialization {
if (!registeredInRemoteNodeDuringSerialization) {
Actor.log.debug("Register serialized Actor [%s] as remote @ [%s:%s]", actorClass.getName, host, port)
RemoteServer.getOrCreateServer(homeAddress)
- RemoteServer.registerActor(homeAddress, uuid, ar)
+ RemoteServer.registerActorByUuid(homeAddress, uuid.toString, ar)
registeredInRemoteNodeDuringSerialization = true
}
@@ -239,13 +265,13 @@ object RemoteActorSerialization {
}
def createRemoteRequestProtocolBuilder(
- actorRef: ActorRef,
- message: Any,
- isOneWay: Boolean,
- senderOption: Option[ActorRef],
- typedActorInfo: Option[Tuple2[String, String]],
- actorType: ActorType):
- RemoteRequestProtocol.Builder = {
+ actorRef: ActorRef,
+ message: Any,
+ isOneWay: Boolean,
+ senderOption: Option[ActorRef],
+ typedActorInfo: Option[Tuple2[String, String]],
+ actorType: ActorType):
+ RemoteRequestProtocol.Builder = {
import actorRef._
val actorInfoBuilder = ActorInfoProtocol.newBuilder
@@ -254,12 +280,13 @@ object RemoteActorSerialization {
.setTarget(actorClassName)
.setTimeout(timeout)
- typedActorInfo.foreach { typedActor =>
- actorInfoBuilder.setTypedActorInfo(
- TypedActorInfoProtocol.newBuilder
- .setInterface(typedActor._1)
- .setMethod(typedActor._2)
- .build)
+ typedActorInfo.foreach {
+ typedActor =>
+ actorInfoBuilder.setTypedActorInfo(
+ TypedActorInfoProtocol.newBuilder
+ .setInterface(typedActor._1)
+ .setMethod(typedActor._2)
+ .build)
}
actorType match {
@@ -280,7 +307,107 @@ object RemoteActorSerialization {
senderOption.foreach { sender =>
RemoteServer.getOrCreateServer(sender.homeAddress).register(sender.uuid.toString, sender)
requestBuilder.setSender(toRemoteActorRefProtocol(sender))
+
}
requestBuilder
}
+
+
+}
+
+
+/**
+ * Module for local typed actor serialization.
+ */
+object TypedActorSerialization {
+
+ def fromBinary[T <: Actor, U <: AnyRef](bytes: Array[Byte])(implicit format: Format[T]): U =
+ fromBinaryToLocalTypedActorRef(bytes, format)
+
+ def toBinary[T <: Actor](proxy: AnyRef)(implicit format: Format[T]): Array[Byte] = {
+ toSerializedTypedActorRefProtocol(proxy, format).toByteArray
+ }
+
+ // wrapper for implicits to be used by Java
+ def fromBinaryJ[T <: Actor, U <: AnyRef](bytes: Array[Byte], format: Format[T]): U =
+ fromBinary(bytes)(format)
+
+ // wrapper for implicits to be used by Java
+ def toBinaryJ[T <: Actor](a: AnyRef, format: Format[T]): Array[Byte] =
+ toBinary(a)(format)
+
+ private def toSerializedTypedActorRefProtocol[T <: Actor](
+ proxy: AnyRef, format: Format[T]): SerializedTypedActorRefProtocol = {
+
+ val init = AspectInitRegistry.initFor(proxy)
+ if (init == null) throw new IllegalArgumentException("Proxy for typed actor could not be found in AspectInitRegistry.")
+
+ SerializedTypedActorRefProtocol.newBuilder
+ .setActorRef(ActorSerialization.toSerializedActorRefProtocol(init.actorRef, format))
+ .setInterfaceName(init.interfaceClass.getName)
+ .build
+ }
+
+ private def fromBinaryToLocalTypedActorRef[T <: Actor, U <: AnyRef](bytes: Array[Byte], format: Format[T]): U =
+ fromProtobufToLocalTypedActorRef(SerializedTypedActorRefProtocol.newBuilder.mergeFrom(bytes).build, format, None)
+
+ private def fromProtobufToLocalTypedActorRef[T <: Actor, U <: AnyRef](
+ protocol: SerializedTypedActorRefProtocol, format: Format[T], loader: Option[ClassLoader]): U = {
+ Actor.log.debug("Deserializing SerializedTypedActorRefProtocol to LocalActorRef:\n" + protocol)
+ val actorRef = ActorSerialization.fromProtobufToLocalActorRef(protocol.getActorRef, format, loader)
+ val intfClass = toClass(loader, protocol.getInterfaceName)
+ TypedActor.newInstance(intfClass, actorRef).asInstanceOf[U]
+ }
+
+ private[akka] def toClass[U <: AnyRef](loader: Option[ClassLoader], name: String): Class[U] = {
+ val classLoader = loader.getOrElse(getClass.getClassLoader)
+ val clazz = classLoader.loadClass(name)
+ clazz.asInstanceOf[Class[U]]
+ }
+}
+
+/**
+ * Module for remote typed actor serialization.
+ */
+object RemoteTypedActorSerialization {
+ /**
+ * Deserializes a byte array (Array[Byte]) into an RemoteActorRef instance.
+ */
+ def fromBinaryToRemoteTypedActorRef[T <: AnyRef](bytes: Array[Byte]): T =
+ fromProtobufToRemoteTypedActorRef(RemoteTypedActorRefProtocol.newBuilder.mergeFrom(bytes).build, None)
+
+ /**
+ * Deserializes a byte array (Array[Byte]) into a AW RemoteActorRef proxy.
+ */
+ def fromBinaryToRemoteTypedActorRef[T <: AnyRef](bytes: Array[Byte], loader: ClassLoader): T =
+ fromProtobufToRemoteTypedActorRef(RemoteTypedActorRefProtocol.newBuilder.mergeFrom(bytes).build, Some(loader))
+
+ /**
+ * Serialize as AW RemoteActorRef proxy.
+ */
+ def toBinary[T <: Actor](proxy: AnyRef): Array[Byte] = {
+ toRemoteTypedActorRefProtocol(proxy).toByteArray
+ }
+
+ /**
+ * Deserializes a RemoteTypedActorRefProtocol Protocol Buffers (protobuf) Message into AW RemoteActorRef proxy.
+ */
+ private[akka] def fromProtobufToRemoteTypedActorRef[T](protocol: RemoteTypedActorRefProtocol, loader: Option[ClassLoader]): T = {
+ Actor.log.debug("Deserializing RemoteTypedActorRefProtocol to AW RemoteActorRef proxy:\n" + protocol)
+ val actorRef = RemoteActorSerialization.fromProtobufToRemoteActorRef(protocol.getActorRef, loader)
+ val intfClass = TypedActorSerialization.toClass(loader, protocol.getInterfaceName)
+ TypedActor.createProxyForRemoteActorRef(intfClass, actorRef).asInstanceOf[T]
+ }
+
+ /**
+ * Serializes the AW TypedActor proxy into a Protocol Buffers (protobuf) Message.
+ */
+ def toRemoteTypedActorRefProtocol(proxy: AnyRef): RemoteTypedActorRefProtocol = {
+ val init = AspectInitRegistry.initFor(proxy)
+ RemoteTypedActorRefProtocol.newBuilder
+ .setActorRef(RemoteActorSerialization.toRemoteActorRefProtocol(init.actorRef))
+ .setInterfaceName(init.interfaceClass.getName)
+ .build
+ }
+
}
diff --git a/akka-remote/src/test/scala/remote/ServerInitiatedRemoteActorSpec.scala b/akka-remote/src/test/scala/remote/ServerInitiatedRemoteActorSpec.scala
index 59f122c656..e40d44ad6c 100644
--- a/akka-remote/src/test/scala/remote/ServerInitiatedRemoteActorSpec.scala
+++ b/akka-remote/src/test/scala/remote/ServerInitiatedRemoteActorSpec.scala
@@ -79,7 +79,6 @@ class ServerInitiatedRemoteActorSpec extends JUnitSuite {
}
}
-
@Test
def shouldSendWithBang {
val actor = RemoteClient.actorFor(
@@ -178,5 +177,41 @@ class ServerInitiatedRemoteActorSpec extends JUnitSuite {
assert(actor2.id == actor3.id)
}
+ @Test
+ def shouldFindActorByUuid {
+ val actor1 = actorOf[RemoteActorSpecActorUnidirectional]
+ val actor2 = actorOf[RemoteActorSpecActorUnidirectional]
+ server.register("uuid:" + actor1.uuid, actor1)
+ server.register("my-service", actor2)
+
+ val ref1 = RemoteClient.actorFor("uuid:" + actor1.uuid, HOSTNAME, PORT)
+ val ref2 = RemoteClient.actorFor("my-service", HOSTNAME, PORT)
+
+ ref1 ! "OneWay"
+ assert(RemoteActorSpecActorUnidirectional.latch.await(1, TimeUnit.SECONDS))
+ ref1.stop
+ ref2 ! "OneWay"
+ ref2.stop
+
+ }
+
+ @Test
+ def shouldRegisterAndUnregister {
+ val actor1 = actorOf[RemoteActorSpecActorUnidirectional]
+ server.register("my-service-1", actor1)
+ assert(server.actors().get("my-service-1") != null, "actor registered")
+ server.unregister("my-service-1")
+ assert(server.actors().get("my-service-1") == null, "actor unregistered")
+ }
+
+ @Test
+ def shouldRegisterAndUnregisterByUuid {
+ val actor1 = actorOf[RemoteActorSpecActorUnidirectional]
+ server.register("uuid:" + actor1.uuid, actor1)
+ assert(server.actorsByUuid().get(actor1.uuid.toString) != null, "actor registered")
+ server.unregister("uuid:" + actor1.uuid)
+ assert(server.actorsByUuid().get(actor1.uuid) == null, "actor unregistered")
+ }
+
}
diff --git a/akka-remote/src/test/scala/remote/ServerInitiatedRemoteTypedActorSpec.scala b/akka-remote/src/test/scala/remote/ServerInitiatedRemoteTypedActorSpec.scala
index b800fbf2c3..71ece9792e 100644
--- a/akka-remote/src/test/scala/remote/ServerInitiatedRemoteTypedActorSpec.scala
+++ b/akka-remote/src/test/scala/remote/ServerInitiatedRemoteTypedActorSpec.scala
@@ -103,9 +103,34 @@ class ServerInitiatedRemoteTypedActorSpec extends
it("should register and unregister typed actors") {
val typedActor = TypedActor.newInstance(classOf[RemoteTypedActorOne], classOf[RemoteTypedActorOneImpl], 1000)
server.registerTypedActor("my-test-service", typedActor)
- assert(server.typedActors().get("my-test-service") != null)
+ assert(server.typedActors().get("my-test-service") ne null, "typed actor registered")
server.unregisterTypedActor("my-test-service")
- assert(server.typedActors().get("my-test-service") == null)
+ assert(server.typedActors().get("my-test-service") eq null, "typed actor unregistered")
+ }
+
+ it("should register and unregister typed actors by uuid") {
+ val typedActor = TypedActor.newInstance(classOf[RemoteTypedActorOne], classOf[RemoteTypedActorOneImpl], 1000)
+ val init = AspectInitRegistry.initFor(typedActor)
+ val uuid = "uuid:" + init.actorRef.uuid
+ server.registerTypedActor(uuid, typedActor)
+ assert(server.typedActorsByUuid().get(init.actorRef.uuid.toString) ne null, "typed actor registered")
+ server.unregisterTypedActor(uuid)
+ assert(server.typedActorsByUuid().get(init.actorRef.uuid.toString) eq null, "typed actor unregistered")
+ }
+
+ it("should find typed actors by uuid") {
+ val typedActor = TypedActor.newInstance(classOf[RemoteTypedActorOne], classOf[RemoteTypedActorOneImpl], 1000)
+ val init = AspectInitRegistry.initFor(typedActor)
+ val uuid = "uuid:" + init.actorRef.uuid
+ server.registerTypedActor(uuid, typedActor)
+ assert(server.typedActorsByUuid().get(init.actorRef.uuid.toString) ne null, "typed actor registered")
+
+ val actor = RemoteClient.typedActorFor(classOf[RemoteTypedActorOne], uuid, HOSTNAME, PORT)
+ expect("oneway") {
+ actor.oneWay
+ oneWayLog.poll(5, TimeUnit.SECONDS)
+ }
+
}
}
}
diff --git a/akka-remote/src/test/scala/serialization/SerializableTypeClassActorSpec.scala b/akka-remote/src/test/scala/serialization/SerializableTypeClassActorSpec.scala
index 7e8babe168..832a655c22 100644
--- a/akka-remote/src/test/scala/serialization/SerializableTypeClassActorSpec.scala
+++ b/akka-remote/src/test/scala/serialization/SerializableTypeClassActorSpec.scala
@@ -127,9 +127,16 @@ class SerializableTypeClassActorSpec extends
(actor1 ! "hello")
(actor1 ! "hello")
(actor1 ! "hello")
+ actor1.mailboxSize should be > (0)
val actor2 = fromBinary(toBinary(actor1))
Thread.sleep(1000)
+ actor2.mailboxSize should be > (0)
(actor2 !! "hello-reply").getOrElse("_") should equal("world")
+
+ val actor3 = fromBinary(toBinary(actor1, false))
+ Thread.sleep(1000)
+ actor3.mailboxSize should equal(0)
+ (actor3 !! "hello-reply").getOrElse("_") should equal("world")
}
}
}
diff --git a/akka-remote/src/test/scala/serialization/Ticket435Spec.scala b/akka-remote/src/test/scala/serialization/Ticket435Spec.scala
new file mode 100644
index 0000000000..ed175ea0ad
--- /dev/null
+++ b/akka-remote/src/test/scala/serialization/Ticket435Spec.scala
@@ -0,0 +1,126 @@
+package se.scalablesolutions.akka.actor.serialization
+
+
+import org.scalatest.Spec
+import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.junit.JUnitRunner
+import org.junit.runner.RunWith
+
+import se.scalablesolutions.akka.serialization._
+import se.scalablesolutions.akka.actor._
+import ActorSerialization._
+import Actor._
+
+@RunWith(classOf[JUnitRunner])
+class Ticket435Spec extends
+ Spec with
+ ShouldMatchers with
+ BeforeAndAfterAll {
+
+ object BinaryFormatMyStatefulActor {
+ implicit object MyStatefulActorFormat extends Format[MyStatefulActor] {
+ def fromBinary(bytes: Array[Byte], act: MyStatefulActor) = {
+ val p = Serializer.Protobuf.fromBinary(bytes, Some(classOf[ProtobufProtocol.Counter])).asInstanceOf[ProtobufProtocol.Counter]
+ act.count = p.getCount
+ act
+ }
+ def toBinary(ac: MyStatefulActor) =
+ ProtobufProtocol.Counter.newBuilder.setCount(ac.count).build.toByteArray
+ }
+ }
+
+ object BinaryFormatMyStatelessActorWithMessagesInMailbox {
+ implicit object MyStatelessActorFormat extends StatelessActorFormat[MyStatelessActorWithMessagesInMailbox]
+ }
+
+ describe("Serializable actor") {
+
+ it("should be able to serialize and deserialize a stateless actor with messages in mailbox") {
+ import BinaryFormatMyStatelessActorWithMessagesInMailbox._
+
+ val actor1 = actorOf[MyStatelessActorWithMessagesInMailbox].start
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ actor1.mailboxSize should be > (0)
+ val actor2 = fromBinary(toBinary(actor1))
+ Thread.sleep(1000)
+ actor2.mailboxSize should be > (0)
+ (actor2 !! "hello-reply").getOrElse("_") should equal("world")
+
+ val actor3 = fromBinary(toBinary(actor1, false))
+ Thread.sleep(1000)
+ actor3.mailboxSize should equal(0)
+ (actor3 !! "hello-reply").getOrElse("_") should equal("world")
+ }
+
+ it("should serialize the mailbox optionally") {
+ import BinaryFormatMyStatelessActorWithMessagesInMailbox._
+
+ val actor1 = actorOf[MyStatelessActorWithMessagesInMailbox].start
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ (actor1 ! "hello")
+ actor1.mailboxSize should be > (0)
+
+ val actor2 = fromBinary(toBinary(actor1, false))
+ Thread.sleep(1000)
+ actor2.mailboxSize should equal(0)
+ (actor2 !! "hello-reply").getOrElse("_") should equal("world")
+ }
+
+ it("should be able to serialize and deserialize a stateful actor with messages in mailbox") {
+ import BinaryFormatMyStatefulActor._
+
+ val actor1 = actorOf[MyStatefulActor].start
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ (actor1 ! "hi")
+ actor1.mailboxSize should be > (0)
+ val actor2 = fromBinary(toBinary(actor1))
+ Thread.sleep(1000)
+ actor2.mailboxSize should be > (0)
+ (actor2 !! "hello").getOrElse("_") should equal("world 1")
+
+ val actor3 = fromBinary(toBinary(actor1, false))
+ Thread.sleep(1000)
+ actor3.mailboxSize should equal(0)
+ (actor3 !! "hello").getOrElse("_") should equal("world 1")
+ }
+ }
+}
+
+class MyStatefulActor extends Actor {
+ var count = 0
+
+ def receive = {
+ case "hi" =>
+ println("# messages in mailbox " + self.mailboxSize)
+ Thread.sleep(500)
+ case "hello" =>
+ count = count + 1
+ self.reply("world " + count)
+ }
+}
diff --git a/akka-remote/src/test/scala/serialization/TypedActorSerializationSpec.scala b/akka-remote/src/test/scala/serialization/TypedActorSerializationSpec.scala
new file mode 100644
index 0000000000..ccf4d05f7f
--- /dev/null
+++ b/akka-remote/src/test/scala/serialization/TypedActorSerializationSpec.scala
@@ -0,0 +1,166 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB
+ */
+package se.scalablesolutions.akka.actor.serialization
+
+import org.scalatest.Spec
+import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.junit.JUnitRunner
+import org.junit.runner.RunWith
+
+import se.scalablesolutions.akka.serialization._
+import se.scalablesolutions.akka.actor._
+
+import TypedActorSerialization._
+import Actor._
+import se.scalablesolutions.akka.remote.{RemoteClient, RemoteServer}
+import se.scalablesolutions.akka.actor.remote.ServerInitiatedRemoteActorSpec.RemoteActorSpecActorUnidirectional
+
+@RunWith(classOf[JUnitRunner])
+class TypedActorSerializationSpec extends
+ Spec with
+ ShouldMatchers with
+ BeforeAndAfterAll {
+
+ var server1: RemoteServer = null
+ var typedActor: MyTypedActor = null
+
+ override def beforeAll = {
+ server1 = new RemoteServer().start("localhost", 9991)
+ typedActor = TypedActor.newInstance(classOf[MyTypedActor], classOf[MyTypedActorImpl], 1000)
+ server1.registerTypedActor("typed-actor-service", typedActor)
+ Thread.sleep(1000)
+ }
+
+ // make sure the servers shutdown cleanly after the test has finished
+ override def afterAll = {
+ try {
+ TypedActor.stop(typedActor)
+ server1.shutdown
+ RemoteClient.shutdownAll
+ Thread.sleep(1000)
+ } catch {
+ case e => ()
+ }
+ }
+
+ object MyTypedStatelessActorFormat extends StatelessActorFormat[MyStatelessTypedActorImpl]
+
+ class MyTypedActorFormat extends Format[MyTypedActorImpl] {
+ def fromBinary(bytes: Array[Byte], act: MyTypedActorImpl) = {
+ val p = Serializer.Protobuf.fromBinary(bytes, Some(classOf[ProtobufProtocol.Counter])).asInstanceOf[ProtobufProtocol.Counter]
+ act.count = p.getCount
+ act
+ }
+ def toBinary(ac: MyTypedActorImpl) =
+ ProtobufProtocol.Counter.newBuilder.setCount(ac.count).build.toByteArray
+ }
+
+ class MyTypedActorWithDualCounterFormat extends Format[MyTypedActorWithDualCounter] {
+ def fromBinary(bytes: Array[Byte], act: MyTypedActorWithDualCounter) = {
+ val p = Serializer.Protobuf.fromBinary(bytes, Some(classOf[ProtobufProtocol.DualCounter])).asInstanceOf[ProtobufProtocol.DualCounter]
+ act.count1 = p.getCount1
+ act.count2 = p.getCount2
+ act
+ }
+ def toBinary(ac: MyTypedActorWithDualCounter) =
+ ProtobufProtocol.DualCounter.newBuilder.setCount1(ac.count1).setCount2(ac.count2).build.toByteArray
+ }
+
+
+ describe("Serializable typed actor") {
+
+ it("should be able to serialize and de-serialize a stateless typed actor") {
+ val typedActor1 = TypedActor.newInstance(classOf[MyTypedActor], classOf[MyStatelessTypedActorImpl], 1000)
+ typedActor1.requestReply("hello") should equal("world")
+ typedActor1.requestReply("hello") should equal("world")
+
+ val bytes = toBinaryJ(typedActor1, MyTypedStatelessActorFormat)
+ val typedActor2: MyTypedActor = fromBinaryJ(bytes, MyTypedStatelessActorFormat)
+ typedActor2.requestReply("hello") should equal("world")
+ }
+
+ it("should be able to serialize and de-serialize a stateful typed actor") {
+ val typedActor1 = TypedActor.newInstance(classOf[MyTypedActor], classOf[MyTypedActorImpl], 1000)
+ typedActor1.requestReply("hello") should equal("world 1")
+ typedActor1.requestReply("scala") should equal("hello scala 2")
+
+ val f = new MyTypedActorFormat
+ val bytes = toBinaryJ(typedActor1, f)
+ val typedActor2: MyTypedActor = fromBinaryJ(bytes, f)
+ typedActor2.requestReply("hello") should equal("world 3")
+ }
+
+ it("should be able to serialize and de-serialize a stateful typed actor with compound state") {
+ val typedActor1 = TypedActor.newInstance(classOf[MyTypedActor], classOf[MyTypedActorWithDualCounter], 1000)
+ typedActor1.requestReply("hello") should equal("world 1 1")
+ typedActor1.requestReply("hello") should equal("world 2 2")
+
+ val f = new MyTypedActorWithDualCounterFormat
+ val bytes = toBinaryJ(typedActor1, f)
+ val typedActor2: MyTypedActor = fromBinaryJ(bytes, f)
+ typedActor2.requestReply("hello") should equal("world 3 3")
+ }
+
+ it("should be able to serialize a local yped actor ref to a remote typed actor ref proxy") {
+ val typedActor1 = TypedActor.newInstance(classOf[MyTypedActor], classOf[MyStatelessTypedActorImpl], 1000)
+ typedActor1.requestReply("hello") should equal("world")
+ typedActor1.requestReply("hello") should equal("world")
+
+ val bytes = RemoteTypedActorSerialization.toBinary(typedActor1)
+ val typedActor2: MyTypedActor = RemoteTypedActorSerialization.fromBinaryToRemoteTypedActorRef(bytes)
+ typedActor1.requestReply("hello") should equal("world")
+ }
+ }
+}
+
+
+trait MyTypedActor {
+ def requestReply(s: String) : String
+ def oneWay() : Unit
+}
+
+class MyTypedActorImpl extends TypedActor with MyTypedActor {
+ var count = 0
+
+ override def oneWay() {
+ println("got oneWay message")
+ }
+
+ override def requestReply(message: String) : String = {
+ count = count + 1
+ if (message == "hello") {
+ "world " + count
+ } else ("hello " + message + " " + count)
+ }
+}
+
+class MyTypedActorWithDualCounter extends TypedActor with MyTypedActor {
+ var count1 = 0
+ var count2 = 0
+
+ override def oneWay() {
+ println("got oneWay message")
+ }
+
+ override def requestReply(message: String) : String = {
+ count1 = count1 + 1
+ count2 = count2 + 1
+
+ if (message == "hello") {
+ "world " + count1 + " " + count2
+ } else ("hello " + message + " " + count1 + " " + count2)
+ }
+}
+
+class MyStatelessTypedActorImpl extends TypedActor with MyTypedActor {
+
+ override def oneWay() {
+ println("got oneWay message")
+ }
+
+ override def requestReply(message: String) : String = {
+ if (message == "hello") "world" else ("hello " + message)
+ }
+}
diff --git a/akka-typed-actor/src/main/scala/actor/TypedActor.scala b/akka-typed-actor/src/main/scala/actor/TypedActor.scala
index 2ae01a5670..a4c7ddada1 100644
--- a/akka-typed-actor/src/main/scala/actor/TypedActor.scala
+++ b/akka-typed-actor/src/main/scala/actor/TypedActor.scala
@@ -390,11 +390,22 @@ object TypedActor extends Logging {
if (config._messageDispatcher.isDefined) actorRef.dispatcher = config._messageDispatcher.get
if (config._threadBasedDispatcher.isDefined) actorRef.dispatcher = Dispatchers.newThreadBasedDispatcher(actorRef)
if (config._host.isDefined) actorRef.makeRemote(config._host.get)
+ actorRef.timeout = config.timeout
AspectInitRegistry.register(proxy, AspectInit(intfClass, typedActor, actorRef, config._host, config.timeout))
actorRef.start
proxy.asInstanceOf[T]
}
+ private[akka] def newInstance[T](intfClass: Class[T], actorRef: ActorRef): T = {
+ if (!actorRef.actorInstance.get.isInstanceOf[TypedActor]) throw new IllegalArgumentException("ActorRef is not a ref to a typed actor")
+ val typedActor = actorRef.actorInstance.get.asInstanceOf[TypedActor]
+ val proxy = Proxy.newInstance(Array(intfClass), Array(typedActor), true, false)
+ typedActor.initialize(proxy)
+ AspectInitRegistry.register(proxy, AspectInit(intfClass, typedActor, actorRef, actorRef.remoteAddress, actorRef.timeout))
+ actorRef.start
+ proxy.asInstanceOf[T]
+ }
+
private[akka] def newInstance[T](intfClass: Class[T], targetClass: Class[_],
remoteAddress: Option[InetSocketAddress], timeout: Long): T = {
val actorRef = actorOf(newTypedActor(targetClass))