raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Removed cross layer dependencies

Browse source 
- akka protocol treats upper layer messages as raw bytes now
 - deserialization for received messages no longer use localAddressForRemote
This commit is contained in:
Endre Sándor Varga 2012-11-21 15:58:01 +01:00
parent 682abccf7e
commit d03d98b6ab
10 changed files with 138 additions and 211 deletions
Download patch file Download diff file Expand all files Collapse all files

225
akka-remote/src/main/java/akka/remote/RemoteProtocol.java
View file

@ -21,7 +21,7 @@ public final class RemoteProtocol {
public final int getNumber() { return value; } public final int getNumber() { return value; }
public static CommandType valueOf(int value) { public static CommandType valueOf(int value) {
switch (value) { switch (value) {
case 1: return CONNECT; case 1: return CONNECT;
@ -83,10 +83,9 @@ public final class RemoteProtocol {
public interface AkkaRemoteProtocolOrBuilder public interface AkkaRemoteProtocolOrBuilder
extends com.google.protobuf.MessageOrBuilder { extends com.google.protobuf.MessageOrBuilder {
// optional .RemoteMessageProtocol message = 1; // optional bytes payload = 1;
boolean hasMessage(); boolean hasPayload();
akka.remote.RemoteProtocol.RemoteMessageProtocol getMessage(); com.google.protobuf.ByteString getPayload();
akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder getMessageOrBuilder();
// optional .RemoteControlProtocol instruction = 2; // optional .RemoteControlProtocol instruction = 2;
boolean hasInstruction(); boolean hasInstruction();
@ -122,17 +121,14 @@ public final class RemoteProtocol {
} }
private int bitField0_; private int bitField0_;
// optional .RemoteMessageProtocol message = 1; // optional bytes payload = 1;
public static final int MESSAGE_FIELD_NUMBER = 1; public static final int PAYLOAD_FIELD_NUMBER = 1;
private akka.remote.RemoteProtocol.RemoteMessageProtocol message_; private com.google.protobuf.ByteString payload_;
public boolean hasMessage() { public boolean hasPayload() {
return ((bitField0_ & 0x00000001) == 0x00000001); return ((bitField0_ & 0x00000001) == 0x00000001);
} }
public akka.remote.RemoteProtocol.RemoteMessageProtocol getMessage() { public com.google.protobuf.ByteString getPayload() {
return message_; return payload_;
}
public akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder getMessageOrBuilder() {
return message_;
} }
// optional .RemoteControlProtocol instruction = 2; // optional .RemoteControlProtocol instruction = 2;
@ -149,7 +145,7 @@ public final class RemoteProtocol {
} }
private void initFields() { private void initFields() {
message_ = akka.remote.RemoteProtocol.RemoteMessageProtocol.getDefaultInstance(); payload_ = com.google.protobuf.ByteString.EMPTY;
instruction_ = akka.remote.RemoteProtocol.RemoteControlProtocol.getDefaultInstance(); instruction_ = akka.remote.RemoteProtocol.RemoteControlProtocol.getDefaultInstance();
} }
private byte memoizedIsInitialized = -1; private byte memoizedIsInitialized = -1;
@ -157,12 +153,6 @@ public final class RemoteProtocol {
byte isInitialized = memoizedIsInitialized; byte isInitialized = memoizedIsInitialized;
if (isInitialized != -1) return isInitialized == 1; if (isInitialized != -1) return isInitialized == 1;
if (hasMessage()) {
if (!getMessage().isInitialized()) {
memoizedIsInitialized = 0;
return false;
}
}
if (hasInstruction()) { if (hasInstruction()) {
if (!getInstruction().isInitialized()) { if (!getInstruction().isInitialized()) {
memoizedIsInitialized = 0; memoizedIsInitialized = 0;
@ -177,7 +167,7 @@ public final class RemoteProtocol {
throws java.io.IOException { throws java.io.IOException {
getSerializedSize(); getSerializedSize();
if (((bitField0_ & 0x00000001) == 0x00000001)) { if (((bitField0_ & 0x00000001) == 0x00000001)) {
output.writeMessage(1, message_); output.writeBytes(1, payload_);
} }
if (((bitField0_ & 0x00000002) == 0x00000002)) { if (((bitField0_ & 0x00000002) == 0x00000002)) {
output.writeMessage(2, instruction_); output.writeMessage(2, instruction_);
@ -193,7 +183,7 @@ public final class RemoteProtocol {
size = 0; size = 0;
if (((bitField0_ & 0x00000001) == 0x00000001)) { if (((bitField0_ & 0x00000001) == 0x00000001)) {
size += com.google.protobuf.CodedOutputStream size += com.google.protobuf.CodedOutputStream
.computeMessageSize(1, message_); .computeBytesSize(1, payload_);
} }
if (((bitField0_ & 0x00000002) == 0x00000002)) { if (((bitField0_ & 0x00000002) == 0x00000002)) {
size += com.google.protobuf.CodedOutputStream size += com.google.protobuf.CodedOutputStream
@ -315,7 +305,6 @@ public final class RemoteProtocol {
} }
private void maybeForceBuilderInitialization() { private void maybeForceBuilderInitialization() {
if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) {
getMessageFieldBuilder();
getInstructionFieldBuilder(); getInstructionFieldBuilder();
} }
} }
@ -325,11 +314,7 @@ public final class RemoteProtocol {
public Builder clear() { public Builder clear() {
super.clear(); super.clear();
if (messageBuilder_ == null) { payload_ = com.google.protobuf.ByteString.EMPTY;
message_ = akka.remote.RemoteProtocol.RemoteMessageProtocol.getDefaultInstance();
} else {
messageBuilder_.clear();
}
bitField0_ = (bitField0_ & ~0x00000001); bitField0_ = (bitField0_ & ~0x00000001);
if (instructionBuilder_ == null) { if (instructionBuilder_ == null) {
instruction_ = akka.remote.RemoteProtocol.RemoteControlProtocol.getDefaultInstance(); instruction_ = akka.remote.RemoteProtocol.RemoteControlProtocol.getDefaultInstance();
@ -378,11 +363,7 @@ public final class RemoteProtocol {
if (((from_bitField0_ & 0x00000001) == 0x00000001)) { if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
to_bitField0_ |= 0x00000001; to_bitField0_ |= 0x00000001;
} }
if (messageBuilder_ == null) { result.payload_ = payload_;
result.message_ = message_;
} else {
result.message_ = messageBuilder_.build();
}
if (((from_bitField0_ & 0x00000002) == 0x00000002)) { if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
to_bitField0_ |= 0x00000002; to_bitField0_ |= 0x00000002;
} }
@ -407,8 +388,8 @@ public final class RemoteProtocol {
public Builder mergeFrom(akka.remote.RemoteProtocol.AkkaRemoteProtocol other) { public Builder mergeFrom(akka.remote.RemoteProtocol.AkkaRemoteProtocol other) {
if (other == akka.remote.RemoteProtocol.AkkaRemoteProtocol.getDefaultInstance()) return this; if (other == akka.remote.RemoteProtocol.AkkaRemoteProtocol.getDefaultInstance()) return this;
if (other.hasMessage()) { if (other.hasPayload()) {
mergeMessage(other.getMessage()); setPayload(other.getPayload());
} }
if (other.hasInstruction()) { if (other.hasInstruction()) {
mergeInstruction(other.getInstruction()); mergeInstruction(other.getInstruction());
@ -418,12 +399,6 @@ public final class RemoteProtocol {
} }
public final boolean isInitialized() { public final boolean isInitialized() {
if (hasMessage()) {
if (!getMessage().isInitialized()) {
return false;
}
}
if (hasInstruction()) { if (hasInstruction()) {
if (!getInstruction().isInitialized()) { if (!getInstruction().isInitialized()) {
@ -457,12 +432,8 @@ public final class RemoteProtocol {
break; break;
} }
case 10: { case 10: {
akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder subBuilder = akka.remote.RemoteProtocol.RemoteMessageProtocol.newBuilder(); bitField0_ |= 0x00000001;
if (hasMessage()) { payload_ = input.readBytes();
subBuilder.mergeFrom(getMessage());
}
input.readMessage(subBuilder, extensionRegistry);
setMessage(subBuilder.buildPartial());
break; break;
} }
case 18: { case 18: {
@ -480,94 +451,28 @@ public final class RemoteProtocol {
private int bitField0_; private int bitField0_;
// optional .RemoteMessageProtocol message = 1; // optional bytes payload = 1;
private akka.remote.RemoteProtocol.RemoteMessageProtocol message_ = akka.remote.RemoteProtocol.RemoteMessageProtocol.getDefaultInstance(); private com.google.protobuf.ByteString payload_ = com.google.protobuf.ByteString.EMPTY;
private com.google.protobuf.SingleFieldBuilder< public boolean hasPayload() {
akka.remote.RemoteProtocol.RemoteMessageProtocol, akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder, akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder> messageBuilder_;
public boolean hasMessage() {
return ((bitField0_ & 0x00000001) == 0x00000001); return ((bitField0_ & 0x00000001) == 0x00000001);
} }
public akka.remote.RemoteProtocol.RemoteMessageProtocol getMessage() { public com.google.protobuf.ByteString getPayload() {
if (messageBuilder_ == null) { return payload_;
return message_;
} else {
return messageBuilder_.getMessage();
}
} }
public Builder setMessage(akka.remote.RemoteProtocol.RemoteMessageProtocol value) { public Builder setPayload(com.google.protobuf.ByteString value) {
if (messageBuilder_ == null) { if (value == null) {
if (value == null) { throw new NullPointerException();
throw new NullPointerException(); }
} bitField0_ |= 0x00000001;
message_ = value; payload_ = value;
onChanged();
} else {
messageBuilder_.setMessage(value);
}
bitField0_ |= 0x00000001;
return this;
}
public Builder setMessage(
akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder builderForValue) {
if (messageBuilder_ == null) {
message_ = builderForValue.build();
onChanged();
} else {
messageBuilder_.setMessage(builderForValue.build());
}
bitField0_ |= 0x00000001;
return this;
}
public Builder mergeMessage(akka.remote.RemoteProtocol.RemoteMessageProtocol value) {
if (messageBuilder_ == null) {
if (((bitField0_ & 0x00000001) == 0x00000001) &&
message_ != akka.remote.RemoteProtocol.RemoteMessageProtocol.getDefaultInstance()) {
message_ =
akka.remote.RemoteProtocol.RemoteMessageProtocol.newBuilder(message_).mergeFrom(value).buildPartial();
} else {
message_ = value;
}
onChanged();
} else {
messageBuilder_.mergeFrom(value);
}
bitField0_ |= 0x00000001;
return this;
}
public Builder clearMessage() {
if (messageBuilder_ == null) {
message_ = akka.remote.RemoteProtocol.RemoteMessageProtocol.getDefaultInstance();
onChanged();
} else {
messageBuilder_.clear();
}
bitField0_ = (bitField0_ & ~0x00000001);
return this;
}
public akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder getMessageBuilder() {
bitField0_ |= 0x00000001;
onChanged(); onChanged();
return getMessageFieldBuilder().getBuilder(); return this;
} }
public akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder getMessageOrBuilder() { public Builder clearPayload() {
if (messageBuilder_ != null) { bitField0_ = (bitField0_ & ~0x00000001);
return messageBuilder_.getMessageOrBuilder(); payload_ = getDefaultInstance().getPayload();
} else { onChanged();
return message_; return this;
}
}
private com.google.protobuf.SingleFieldBuilder<
akka.remote.RemoteProtocol.RemoteMessageProtocol, akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder, akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder>
getMessageFieldBuilder() {
if (messageBuilder_ == null) {
messageBuilder_ = new com.google.protobuf.SingleFieldBuilder<
akka.remote.RemoteProtocol.RemoteMessageProtocol, akka.remote.RemoteProtocol.RemoteMessageProtocol.Builder, akka.remote.RemoteProtocol.RemoteMessageProtocolOrBuilder>(
message_,
getParentForChildren(),
isClean());
message_ = null;
}
return messageBuilder_;
} }
// optional .RemoteControlProtocol instruction = 2; // optional .RemoteControlProtocol instruction = 2;
@ -6504,34 +6409,34 @@ public final class RemoteProtocol {
descriptor; descriptor;
static { static {
java.lang.String[] descriptorData = { java.lang.String[] descriptorData = {
"\n\024RemoteProtocol.proto\"j\n\022AkkaRemoteProt" + "\n\024RemoteProtocol.proto\"R\n\022AkkaRemoteProt" +
"ocol\022\'\n\007message\030\001 \001(\0132\026.RemoteMessagePro" + "ocol\022\017\n\007payload\030\001 \001(\014\022+\n\013instruction\030\002 \001" +
"tocol\022+\n\013instruction\030\002 \001(\0132\026.RemoteContr" + "(\0132\026.RemoteControlProtocol\"\255\001\n\025RemoteMes" +
"olProtocol\"\255\001\n\025RemoteMessageProtocol\022$\n\t" + "sageProtocol\022$\n\trecipient\030\001 \002(\0132\021.ActorR" +
"recipient\030\001 \002(\0132\021.ActorRefProtocol\022!\n\007me" + "efProtocol\022!\n\007message\030\002 \002(\0132\020.MessagePro" +
"ssage\030\002 \002(\0132\020.MessageProtocol\022!\n\006sender\030" + "tocol\022!\n\006sender\030\004 \001(\0132\021.ActorRefProtocol" +
"\004 \001(\0132\021.ActorRefProtocol\022(\n\010metadata\030\005 \003" + "\022(\n\010metadata\030\005 \003(\0132\026.MetadataEntryProtoc" +
"(\0132\026.MetadataEntryProtocol\"l\n\025RemoteCont" + "ol\"l\n\025RemoteControlProtocol\022!\n\013commandTy" +
"rolProtocol\022!\n\013commandType\030\001 \002(\0162\014.Comma" + "pe\030\001 \002(\0162\014.CommandType\022\016\n\006cookie\030\002 \001(\t\022 " +
"ndType\022\016\n\006cookie\030\002 \001(\t\022 \n\006origin\030\003 \001(\0132\020", "\n\006origin\030\003 \001(\0132\020.AddressProtocol\" \n\020Acto",
".AddressProtocol\" \n\020ActorRefProtocol\022\014\n\004" + "rRefProtocol\022\014\n\004path\030\001 \002(\t\"Q\n\017MessagePro" +
"path\030\001 \002(\t\"Q\n\017MessageProtocol\022\017\n\007message" + "tocol\022\017\n\007message\030\001 \002(\014\022\024\n\014serializerId\030\002" +
"\030\001 \002(\014\022\024\n\014serializerId\030\002 \002(\005\022\027\n\017messageM" + " \002(\005\022\027\n\017messageManifest\030\003 \001(\014\"3\n\025Metadat" +
"anifest\030\003 \001(\014\"3\n\025MetadataEntryProtocol\022\013" + "aEntryProtocol\022\013\n\003key\030\001 \002(\t\022\r\n\005value\030\002 \002" +
"\n\003key\030\001 \002(\t\022\r\n\005value\030\002 \002(\014\"S\n\017AddressPro" + "(\014\"S\n\017AddressProtocol\022\016\n\006system\030\001 \002(\t\022\020\n" +
"tocol\022\016\n\006system\030\001 \002(\t\022\020\n\010hostname\030\002 \002(\t\022" + "\010hostname\030\002 \002(\t\022\014\n\004port\030\003 \002(\r\022\020\n\010protoco" +
"\014\n\004port\030\003 \002(\r\022\020\n\010protocol\030\004 \001(\t\"\216\001\n\027Daem" + "l\030\004 \001(\t\"\216\001\n\027DaemonMsgCreateProtocol\022\035\n\005p" +
"onMsgCreateProtocol\022\035\n\005props\030\001 \002(\0132\016.Pro" + "rops\030\001 \002(\0132\016.PropsProtocol\022\037\n\006deploy\030\002 \002" +
"psProtocol\022\037\n\006deploy\030\002 \002(\0132\017.DeployProto" + "(\0132\017.DeployProtocol\022\014\n\004path\030\003 \002(\t\022%\n\nsup" +
"col\022\014\n\004path\030\003 \002(\t\022%\n\nsupervisor\030\004 \002(\0132\021.", "ervisor\030\004 \002(\0132\021.ActorRefProtocol\"\205\001\n\rPro",
"ActorRefProtocol\"\205\001\n\rPropsProtocol\022\022\n\ndi" + "psProtocol\022\022\n\ndispatcher\030\001 \002(\t\022\037\n\006deploy" +
"spatcher\030\001 \002(\t\022\037\n\006deploy\030\002 \002(\0132\017.DeployP" + "\030\002 \002(\0132\017.DeployProtocol\022\030\n\020fromClassCrea" +
"rotocol\022\030\n\020fromClassCreator\030\003 \001(\t\022\017\n\007cre" + "tor\030\003 \001(\t\022\017\n\007creator\030\004 \001(\014\022\024\n\014routerConf" +
"ator\030\004 \001(\014\022\024\n\014routerConfig\030\005 \001(\014\"S\n\016Depl" + "ig\030\005 \001(\014\"S\n\016DeployProtocol\022\014\n\004path\030\001 \002(\t" +
"oyProtocol\022\014\n\004path\030\001 \002(\t\022\016\n\006config\030\002 \001(\014" + "\022\016\n\006config\030\002 \001(\014\022\024\n\014routerConfig\030\003 \001(\014\022\r" +
"\022\024\n\014routerConfig\030\003 \001(\014\022\r\n\005scope\030\004 \001(\014*7\n" + "\n\005scope\030\004 \001(\014*7\n\013CommandType\022\013\n\007CONNECT\020" +
"\013CommandType\022\013\n\007CONNECT\020\001\022\014\n\010SHUTDOWN\020\002\022" + "\001\022\014\n\010SHUTDOWN\020\002\022\r\n\tHEARTBEAT\020\003B\017\n\013akka.r" +
"\r\n\tHEARTBEAT\020\003B\017\n\013akka.remoteH\001" "emoteH\001"
}; };
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@ -6543,7 +6448,7 @@ public final class RemoteProtocol {
internal_static_AkkaRemoteProtocol_fieldAccessorTable = new internal_static_AkkaRemoteProtocol_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable( com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_AkkaRemoteProtocol_descriptor, internal_static_AkkaRemoteProtocol_descriptor,
new java.lang.String[] { "Message", "Instruction", }, new java.lang.String[] { "Payload", "Instruction", },
akka.remote.RemoteProtocol.AkkaRemoteProtocol.class, akka.remote.RemoteProtocol.AkkaRemoteProtocol.class,
akka.remote.RemoteProtocol.AkkaRemoteProtocol.Builder.class); akka.remote.RemoteProtocol.AkkaRemoteProtocol.Builder.class);
internal_static_RemoteMessageProtocol_descriptor = internal_static_RemoteMessageProtocol_descriptor =

2
akka-remote/src/main/protocol/RemoteProtocol.proto
View file

@ -14,7 +14,7 @@ option optimize_for = SPEED;
*******************************************/ *******************************************/
message AkkaRemoteProtocol { message AkkaRemoteProtocol {
optional RemoteMessageProtocol message = 1; optional bytes payload = 1;
optional RemoteControlProtocol instruction = 2; optional RemoteControlProtocol instruction = 2;
} }

20
akka-remote/src/main/scala/akka/remote/Endpoint.scala
View file

@ -173,7 +173,7 @@ private[remote] class EndpointWriter(
when(Writing) { when(Writing) {
case Event(Send(msg, senderOption, recipient), _) ⇒ case Event(Send(msg, senderOption, recipient), _) ⇒
val pdu = codec.constructMessagePdu(recipient.localAddressToUse, recipient, serializeMessage(msg), senderOption) val pdu = codec.constructMessage(recipient.localAddressToUse, recipient, serializeMessage(msg), senderOption)
val success = try handle.write(pdu) catch { val success = try handle.write(pdu) catch {
case NonFatal(e) ⇒ publishAndThrow("Failed to write message to the transport", e) case NonFatal(e) ⇒ publishAndThrow("Failed to write message to the transport", e)
} }
@ -205,7 +205,7 @@ private[remote] class EndpointWriter(
} }
private def startReadEndpoint(): Unit = { private def startReadEndpoint(): Unit = {
reader = context.actorOf(Props(new EndpointReader(codec, msgDispatch)), reader = context.actorOf(Props(new EndpointReader(codec, handle.localAddress, msgDispatch)),
"endpointReader-" + URLEncoder.encode(remoteAddress.toString, "utf-8")) "endpointReader-" + URLEncoder.encode(remoteAddress.toString, "utf-8"))
handle.readHandlerPromise.success(reader) handle.readHandlerPromise.success(reader)
context.watch(reader) context.watch(reader)
@ -221,6 +221,7 @@ private[remote] class EndpointWriter(
private[remote] class EndpointReader( private[remote] class EndpointReader(
val codec: AkkaPduCodec, val codec: AkkaPduCodec,
val localAddress: Address,
val msgDispatch: InboundMessageDispatcher) extends Actor { val msgDispatch: InboundMessageDispatcher) extends Actor {
val provider = context.system.asInstanceOf[ExtendedActorSystem].provider.asInstanceOf[RemoteActorRefProvider] val provider = context.system.asInstanceOf[ExtendedActorSystem].provider.asInstanceOf[RemoteActorRefProvider]
@ -228,18 +229,13 @@ private[remote] class EndpointReader(
override def receive: Receive = { override def receive: Receive = {
case Disassociated ⇒ context.stop(self) case Disassociated ⇒ context.stop(self)
// FIXME: Do 2 step deserialization (old-remoting must be removed first) case InboundPayload(p) ⇒
case InboundPayload(p) ⇒ decodePdu(p) match { val msg = decodePdu(p)
msgDispatch.dispatch(msg.recipient, msg.recipientAddress, msg.serializedMessage, msg.senderOption)
case Message(recipient, recipientAddress, serializedMessage, senderOption) ⇒
msgDispatch.dispatch(recipient, recipientAddress, serializedMessage, senderOption)
case _ ⇒
}
} }
private def decodePdu(pdu: ByteString): AkkaPdu = try { private def decodePdu(pdu: ByteString): Message = try {
codec.decodePdu(pdu, provider) codec.decodeMessage(pdu, provider, localAddress)
} catch { } catch {
case NonFatal(e) ⇒ throw new EndpointException("Error while decoding incoming Akka PDU", e) case NonFatal(e) ⇒ throw new EndpointException("Error while decoding incoming Akka PDU", e)
} }

13
akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
View file

@ -188,6 +188,19 @@ class RemoteActorRefProvider(
case _ ⇒ local.actorFor(ref, path) case _ ⇒ local.actorFor(ref, path)
} }
/*
* INTERNAL API
* Called in deserialization of incoming remote messages. In this case the correct local address is known, therefore
* this method is faster than the actorFor above.
*/
def actorForWithLocalAddress(ref: InternalActorRef, path: String, localAddress: Address): InternalActorRef = path match {
case ActorPathExtractor(address, elems) ⇒
if (isSelfAddress(address)) actorFor(rootGuardian, elems)
else new RemoteActorRef(this, transport, localAddress,
new RootActorPath(address) / elems, Nobody, props = None, deploy = None)
case _ ⇒ local.actorFor(ref, path)
}
def actorFor(ref: InternalActorRef, path: Iterable[String]): InternalActorRef = local.actorFor(ref, path) def actorFor(ref: InternalActorRef, path: Iterable[String]): InternalActorRef = local.actorFor(ref, path)
/** /**

2
akka-remote/src/main/scala/akka/remote/RemoteTransport.scala
View file

@ -239,7 +239,7 @@ abstract class RemoteTransport(val system: ExtendedActorSystem, val provider: Re
* Returns a newly created AkkaRemoteProtocol with the given message payload. * Returns a newly created AkkaRemoteProtocol with the given message payload.
*/ */
def createMessageSendEnvelope(rmp: RemoteMessageProtocol): AkkaRemoteProtocol = def createMessageSendEnvelope(rmp: RemoteMessageProtocol): AkkaRemoteProtocol =
AkkaRemoteProtocol.newBuilder.setMessage(rmp).build AkkaRemoteProtocol.newBuilder.setPayload(rmp.toByteString).build
/** /**
* Returns a newly created AkkaRemoteProtocol with the given control payload. * Returns a newly created AkkaRemoteProtocol with the given control payload.

6
akka-remote/src/main/scala/akka/remote/netty/Client.scala
View file

@ -12,7 +12,7 @@ import org.jboss.netty.channel.{ ChannelFutureListener, ChannelHandler, DefaultC
import org.jboss.netty.handler.codec.frame.{ LengthFieldPrepender, LengthFieldBasedFrameDecoder } import org.jboss.netty.handler.codec.frame.{ LengthFieldPrepender, LengthFieldBasedFrameDecoder }
import org.jboss.netty.handler.execution.ExecutionHandler import org.jboss.netty.handler.execution.ExecutionHandler
import org.jboss.netty.handler.timeout.{ IdleState, IdleStateEvent, IdleStateAwareChannelHandler, IdleStateHandler } import org.jboss.netty.handler.timeout.{ IdleState, IdleStateEvent, IdleStateAwareChannelHandler, IdleStateHandler }
import akka.remote.RemoteProtocol.{ RemoteControlProtocol, CommandType, AkkaRemoteProtocol } import akka.remote.RemoteProtocol.{ RemoteMessageProtocol, RemoteControlProtocol, CommandType, AkkaRemoteProtocol }
import akka.remote.{ RemoteProtocol, RemoteMessage, RemoteLifeCycleEvent, RemoteClientStarted, RemoteClientShutdown, RemoteClientException, RemoteClientError, RemoteClientDisconnected, RemoteClientConnected } import akka.remote.{ RemoteProtocol, RemoteMessage, RemoteLifeCycleEvent, RemoteClientStarted, RemoteClientShutdown, RemoteClientException, RemoteClientError, RemoteClientDisconnected, RemoteClientConnected }
import akka.AkkaException import akka.AkkaException
import akka.event.Logging import akka.event.Logging
@ -273,8 +273,8 @@ private[akka] class ActiveRemoteClientHandler(
case CommandType.SHUTDOWN ⇒ runOnceNow { client.netty.shutdownClientConnection(remoteAddress) } case CommandType.SHUTDOWN ⇒ runOnceNow { client.netty.shutdownClientConnection(remoteAddress) }
case _ ⇒ //Ignore others case _ ⇒ //Ignore others
} }
case arp: AkkaRemoteProtocol if arp.hasMessage ⇒ case arp: AkkaRemoteProtocol if arp.hasPayload ⇒
client.netty.receiveMessage(new RemoteMessage(arp.getMessage, client.netty.system)) client.netty.receiveMessage(new RemoteMessage(RemoteMessageProtocol.parseFrom(arp.getPayload), client.netty.system))
case other ⇒ case other ⇒
throw new RemoteClientException("Unknown message received in remote client handler: " + other, client.netty, client.remoteAddress) throw new RemoteClientException("Unknown message received in remote client handler: " + other, client.netty, client.remoteAddress)
} }

6
akka-remote/src/main/scala/akka/remote/netty/Server.scala
View file

@ -4,7 +4,7 @@
package akka.remote.netty package akka.remote.netty
import akka.actor.Address import akka.actor.Address
import akka.remote.RemoteProtocol.{ RemoteControlProtocol, CommandType, AkkaRemoteProtocol } import akka.remote.RemoteProtocol.{ RemoteMessageProtocol, RemoteControlProtocol, CommandType, AkkaRemoteProtocol }
import akka.remote._ import akka.remote._
import java.net.InetAddress import java.net.InetAddress
import java.net.InetSocketAddress import java.net.InetSocketAddress
@ -146,8 +146,8 @@ private[akka] class RemoteServerHandler(
override def messageReceived(ctx: ChannelHandlerContext, event: MessageEvent) = try { override def messageReceived(ctx: ChannelHandlerContext, event: MessageEvent) = try {
event.getMessage match { event.getMessage match {
case remote: AkkaRemoteProtocol if remote.hasMessage ⇒ case remote: AkkaRemoteProtocol if remote.hasPayload ⇒
netty.receiveMessage(new RemoteMessage(remote.getMessage, netty.system)) netty.receiveMessage(new RemoteMessage(RemoteMessageProtocol.parseFrom(remote.getPayload), netty.system))
case remote: AkkaRemoteProtocol if remote.hasInstruction ⇒ case remote: AkkaRemoteProtocol if remote.hasInstruction ⇒
val instruction = remote.getInstruction val instruction = remote.getInstruction
instruction.getCommandType match { instruction.getCommandType match {

41
akka-remote/src/main/scala/akka/remote/transport/AkkaPduCodec.scala
View file

@ -7,6 +7,7 @@ import akka.remote.transport.AkkaPduCodec._
import akka.remote.{ RemoteActorRefProvider, RemoteProtocol } import akka.remote.{ RemoteActorRefProvider, RemoteProtocol }
import akka.util.ByteString import akka.util.ByteString
import com.google.protobuf.InvalidProtocolBufferException import com.google.protobuf.InvalidProtocolBufferException
import com.google.protobuf.{ ByteString ⇒ PByteString }
class PduCodecException(msg: String, cause: Throwable) extends AkkaException(msg, cause) class PduCodecException(msg: String, cause: Throwable) extends AkkaException(msg, cause)
@ -20,10 +21,12 @@ private[remote] object AkkaPduCodec {
case class Associate(cookie: Option[String], origin: Address) extends AkkaPdu case class Associate(cookie: Option[String], origin: Address) extends AkkaPdu
case object Disassociate extends AkkaPdu case object Disassociate extends AkkaPdu
case object Heartbeat extends AkkaPdu case object Heartbeat extends AkkaPdu
case class Payload(bytes: ByteString) extends AkkaPdu
case class Message(recipient: InternalActorRef, case class Message(recipient: InternalActorRef,
recipientAddress: Address, recipientAddress: Address,
serializedMessage: MessageProtocol, serializedMessage: MessageProtocol,
sender: Option[ActorRef]) extends AkkaPdu senderOption: Option[ActorRef])
} }
/** /**
@ -31,7 +34,9 @@ private[remote] object AkkaPduCodec {
*/ */
private[remote] trait AkkaPduCodec { private[remote] trait AkkaPduCodec {
def constructMessagePdu( def constructPayload(payload: ByteString): ByteString
def constructMessage(
localAddress: Address, localAddress: Address,
recipient: ActorRef, recipient: ActorRef,
serializedMessage: MessageProtocol, serializedMessage: MessageProtocol,
@ -43,13 +48,15 @@ private[remote] trait AkkaPduCodec {
def constructHeartbeat: ByteString def constructHeartbeat: ByteString
def decodePdu(raw: ByteString, provider: RemoteActorRefProvider): AkkaPdu // Effective enough? def decodePdu(raw: ByteString): AkkaPdu
def decodeMessage(raw: ByteString, provider: RemoteActorRefProvider, localAddress: Address): Message
} }
private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec { private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec {
override def constructMessagePdu( override def constructMessage(
localAddress: Address, localAddress: Address,
recipient: ActorRef, recipient: ActorRef,
serializedMessage: MessageProtocol, serializedMessage: MessageProtocol,
@ -61,9 +68,12 @@ private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec {
senderOption foreach { ref ⇒ messageBuilder.setSender(serializeActorRef(localAddress, ref)) } senderOption foreach { ref ⇒ messageBuilder.setSender(serializeActorRef(localAddress, ref)) }
messageBuilder.setMessage(serializedMessage) messageBuilder.setMessage(serializedMessage)
akkaRemoteProtocolToByteString(AkkaRemoteProtocol.newBuilder().setMessage(messageBuilder.build).build) akkaMessageProtocolToByteString(messageBuilder.build)
} }
override def constructPayload(payload: ByteString): ByteString = akkaRemoteProtocolToByteString(
AkkaRemoteProtocol.newBuilder().setPayload(PByteString.copyFrom(payload.asByteBuffer)).build)
override def constructAssociate(cookie: Option[String], origin: Address): ByteString = override def constructAssociate(cookie: Option[String], origin: Address): ByteString =
constructControlMessagePdu(RemoteProtocol.CommandType.CONNECT, cookie, Some(origin)) constructControlMessagePdu(RemoteProtocol.CommandType.CONNECT, cookie, Some(origin))
@ -73,12 +83,12 @@ private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec {
override val constructHeartbeat: ByteString = override val constructHeartbeat: ByteString =
constructControlMessagePdu(RemoteProtocol.CommandType.HEARTBEAT, None, None) constructControlMessagePdu(RemoteProtocol.CommandType.HEARTBEAT, None, None)
override def decodePdu(raw: ByteString, provider: RemoteActorRefProvider): AkkaPdu = { override def decodePdu(raw: ByteString): AkkaPdu = {
try { try {
val pdu = AkkaRemoteProtocol.parseFrom(raw.toArray) val pdu = AkkaRemoteProtocol.parseFrom(raw.toArray)
if (pdu.hasMessage) { if (pdu.hasPayload) {
decodeMessage(pdu.getMessage, provider) Payload(ByteString(pdu.getPayload.asReadOnlyByteBuffer()))
} else if (pdu.hasInstruction) { } else if (pdu.hasInstruction) {
decodeControlPdu(pdu.getInstruction) decodeControlPdu(pdu.getInstruction)
} else { } else {
@ -89,12 +99,18 @@ private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec {
} }
} }
private def decodeMessage(msgPdu: RemoteMessageProtocol, provider: RemoteActorRefProvider): Message = { override def decodeMessage(
raw: ByteString,
provider: RemoteActorRefProvider,
localAddress: Address): Message = {
val msgPdu = RemoteMessageProtocol.parseFrom(raw.toArray)
Message( Message(
recipient = provider.actorFor(provider.rootGuardian, msgPdu.getRecipient.getPath), recipient = provider.actorForWithLocalAddress(provider.rootGuardian, msgPdu.getRecipient.getPath, localAddress),
recipientAddress = AddressFromURIString(msgPdu.getRecipient.getPath), recipientAddress = AddressFromURIString(msgPdu.getRecipient.getPath),
serializedMessage = msgPdu.getMessage, serializedMessage = msgPdu.getMessage,
sender = if (msgPdu.hasSender) Some(provider.actorFor(provider.rootGuardian, msgPdu.getSender.getPath)) else None) senderOption = (if (msgPdu.hasSender)
Some(provider.actorForWithLocalAddress(provider.rootGuardian, msgPdu.getSender.getPath, localAddress))
else None))
} }
private def decodeControlPdu(controlPdu: RemoteControlProtocol): AkkaPdu = { private def decodeControlPdu(controlPdu: RemoteControlProtocol): AkkaPdu = {
@ -128,6 +144,9 @@ private[remote] object AkkaPduProtobufCodec extends AkkaPduCodec {
private def akkaRemoteProtocolToByteString(pdu: AkkaRemoteProtocol): ByteString = ByteString(pdu.toByteArray) private def akkaRemoteProtocolToByteString(pdu: AkkaRemoteProtocol): ByteString = ByteString(pdu.toByteArray)
private def akkaMessageProtocolToByteString(message: RemoteMessageProtocol): ByteString =
ByteString(message.toByteArray)
private def serializeActorRef(defaultAddress: Address, ref: ActorRef): ActorRefProtocol = { private def serializeActorRef(defaultAddress: Address, ref: ActorRef): ActorRefProtocol = {
val fullActorRefString: String = if (ref.path.address.host.isDefined) val fullActorRefString: String = if (ref.path.address.host.isDefined)
ref.path.toString ref.path.toString

21
akka-remote/src/main/scala/akka/remote/transport/AkkaProtocolTransport.scala
View file

@ -224,9 +224,7 @@ private[transport] class AkkaProtocolHandle(
private val codec: AkkaPduCodec) private val codec: AkkaPduCodec)
extends AssociationHandle { extends AssociationHandle {
// FIXME: This is currently a hack! The caller should not know anything about the format of the Akka protocol override def write(payload: ByteString): Boolean = wrappedHandle.write(codec.constructPayload(payload))
// but here it does. This is temporary and will be fixed.
override def write(payload: ByteString): Boolean = wrappedHandle.write(payload)
override def disassociate(): Unit = stateActor ! DisassociateUnderlying override def disassociate(): Unit = stateActor ! DisassociateUnderlying
@ -296,9 +294,6 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
this(InboundUnassociated(associationHandler, wrappedHandle), localAddress, settings, codec, failureDetector) this(InboundUnassociated(associationHandler, wrappedHandle), localAddress, settings, codec, failureDetector)
} }
// FIXME: This may break with ClusterActorRefProvider if it does not extends RemoteActorRefProvider
val provider = context.system.asInstanceOf[ExtendedActorSystem].provider.asInstanceOf[RemoteActorRefProvider]
initialData match { initialData match {
case d: OutboundUnassociated ⇒ case d: OutboundUnassociated ⇒
d.transport.associate(removeScheme(d.remoteAddress)) pipeTo self d.transport.associate(removeScheme(d.remoteAddress)) pipeTo self
@ -375,10 +370,10 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
stop() stop()
// Any other activity is considered an implicit acknowledgement of the association // Any other activity is considered an implicit acknowledgement of the association
case Message(recipient, recipientAddress, serializedMessage, senderOption) ⇒ case Payload(payload) ⇒
sendHeartbeat(wrappedHandle) sendHeartbeat(wrappedHandle)
goto(Open) using goto(Open) using
AssociatedWaitHandler(notifyOutboundHandler(wrappedHandle, statusPromise), wrappedHandle, Queue(p)) AssociatedWaitHandler(notifyOutboundHandler(wrappedHandle, statusPromise), wrappedHandle, Queue(payload))
case Heartbeat ⇒ case Heartbeat ⇒
sendHeartbeat(wrappedHandle) sendHeartbeat(wrappedHandle)
@ -405,9 +400,9 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
case Heartbeat ⇒ failureDetector.heartbeat(); stay() case Heartbeat ⇒ failureDetector.heartbeat(); stay()
case Message(recipient, recipientAddress, serializedMessage, senderOption) ⇒ case Payload(payload) ⇒
// Queue message until handler is registered // Queue message until handler is registered
stay() using AssociatedWaitHandler(handlerFuture, wrappedHandle, queue :+ p) stay() using AssociatedWaitHandler(handlerFuture, wrappedHandle, queue :+ payload)
case _ ⇒ stay() case _ ⇒ stay()
} }
@ -419,8 +414,8 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
case Heartbeat ⇒ failureDetector.heartbeat(); stay() case Heartbeat ⇒ failureDetector.heartbeat(); stay()
case Message(recipient, recipientAddress, serializedMessage, senderOption) ⇒ case Payload(payload) ⇒
handler ! InboundPayload(p) handler ! InboundPayload(payload)
stay() stay()
case _ ⇒ stay() case _ ⇒ stay()
@ -525,7 +520,7 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
} }
private def decodePdu(pdu: ByteString): AkkaPdu = ape("Error while decoding incoming Akka PDU of length: " + pdu.length) { private def decodePdu(pdu: ByteString): AkkaPdu = ape("Error while decoding incoming Akka PDU of length: " + pdu.length) {
codec.decodePdu(pdu, provider) codec.decodePdu(pdu)
} }
// Neither heartbeats neither disassociate cares about backing off if write fails: // Neither heartbeats neither disassociate cares about backing off if write fails:

13
akka-remote/src/test/scala/akka/remote/transport/AkkaProtocolSpec.scala
View file

@ -68,10 +68,9 @@ class AkkaProtocolSpec extends AkkaSpec("""akka.actor.provider = "akka.remote.Re
val codec = AkkaPduProtobufCodec val codec = AkkaPduProtobufCodec
val provider = system.asInstanceOf[ExtendedActorSystem].provider.asInstanceOf[RemoteActorRefProvider]
val testMsg = RemoteProtocol.MessageProtocol.newBuilder().setSerializerId(0).setMessage(PByteString.copyFromUtf8("foo")).build val testMsg = RemoteProtocol.MessageProtocol.newBuilder().setSerializerId(0).setMessage(PByteString.copyFromUtf8("foo")).build
val testMsgPdu: ByteString = codec.constructMessagePdu(localAkkaAddress, self, testMsg, None) val testEnvelope = codec.constructMessage(localAkkaAddress, self, testMsg, None)
val testMsgPdu: ByteString = codec.constructPayload(testEnvelope)
def testHeartbeat = InboundPayload(codec.constructHeartbeat) def testHeartbeat = InboundPayload(codec.constructHeartbeat)
def testPayload = InboundPayload(testMsgPdu) def testPayload = InboundPayload(testMsgPdu)
@ -91,7 +90,7 @@ class AkkaProtocolSpec extends AkkaSpec("""akka.actor.provider = "akka.remote.Re
def lastActivityIsHeartbeat(registry: AssociationRegistry) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match { def lastActivityIsHeartbeat(registry: AssociationRegistry) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match {
case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒ case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒
codec.decodePdu(payload, provider) match { codec.decodePdu(payload) match {
case Heartbeat ⇒ true case Heartbeat ⇒ true
case _ ⇒ false case _ ⇒ false
} }
@ -100,7 +99,7 @@ class AkkaProtocolSpec extends AkkaSpec("""akka.actor.provider = "akka.remote.Re
def lastActivityIsAssociate(registry: AssociationRegistry, cookie: Option[String]) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match { def lastActivityIsAssociate(registry: AssociationRegistry, cookie: Option[String]) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match {
case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒ case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒
codec.decodePdu(payload, provider) match { codec.decodePdu(payload) match {
case Associate(c, origin) if c == cookie && origin == localAddress ⇒ true case Associate(c, origin) if c == cookie && origin == localAddress ⇒ true
case _ ⇒ false case _ ⇒ false
} }
@ -109,7 +108,7 @@ class AkkaProtocolSpec extends AkkaSpec("""akka.actor.provider = "akka.remote.Re
def lastActivityIsDisassociate(registry: AssociationRegistry) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match { def lastActivityIsDisassociate(registry: AssociationRegistry) = if (registry.logSnapshot.isEmpty) false else registry.logSnapshot.last match {
case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒ case WriteAttempt(sender, recipient, payload) if sender == localAddress && recipient == remoteAddress ⇒
codec.decodePdu(payload, provider) match { codec.decodePdu(payload) match {
case Disassociate ⇒ true case Disassociate ⇒ true
case _ ⇒ false case _ ⇒ false
} }
@ -161,7 +160,7 @@ class AkkaProtocolSpec extends AkkaSpec("""akka.actor.provider = "akka.remote.Re
reader ! testPayload reader ! testPayload
expectMsgPF() { expectMsgPF() {
case InboundPayload(p) ⇒ p must be === testMsgPdu case InboundPayload(p) ⇒ p must be === testEnvelope
} }
} }