-
Notifications
You must be signed in to change notification settings - Fork 217
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Replace consistent hashable envelope by a dedicated binary message en…
…velope for communication between the connection persistence actor and its client actors. Reason: Consistent hashable envelope is not serializable. Still to do: Make all messages wrapped in ShardedBinaryEnvelope serializable, i. e., ClientActorPropsArgs. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
- Loading branch information
Showing
8 changed files
with
283 additions
and
22 deletions.
There are no files selected for viewing
35 changes: 35 additions & 0 deletions
35
.../src/main/java/org/eclipse/ditto/connectivity/service/messaging/ClientActorPropsArgs.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| /* | ||
| * Copyright (c) 2022 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.connectivity.service.messaging; | ||
|
|
||
| import javax.annotation.concurrent.Immutable; | ||
|
|
||
| import org.eclipse.ditto.base.model.headers.DittoHeaders; | ||
| import org.eclipse.ditto.connectivity.model.Connection; | ||
|
|
||
| import com.typesafe.config.Config; | ||
|
|
||
| import akka.actor.ActorRef; | ||
|
|
||
| /** | ||
| * Arguments to create a client actor props object. | ||
| * | ||
| * @param connection the connection. | ||
| * @param commandForwarderActor the actor used to send signals into the ditto cluster.. | ||
| * @param connectionActor the connectionPersistenceActor which creates this client. | ||
| * @param dittoHeaders Ditto headers of the command that caused the client actors to be created. | ||
| */ | ||
| @Immutable | ||
| public record ClientActorPropsArgs(Connection connection, ActorRef commandForwarderActor, ActorRef connectionActor, | ||
| DittoHeaders dittoHeaders, Config connectivityConfigOverwrites) { | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
25 changes: 25 additions & 0 deletions
25
...cluster/src/main/java/org/eclipse/ditto/internal/utils/cluster/ShardedBinaryEnvelope.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| /* | ||
| * Copyright (c) 2022 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.internal.utils.cluster; | ||
|
|
||
| import javax.annotation.concurrent.Immutable; | ||
|
|
||
| /** | ||
| * Envelope to hold a binary message to a sharded entity. Serialization is performed recursively. | ||
| * | ||
| * @param message Message to the sharded entity. | ||
| * @param entityName Name of the recipient entity. | ||
| * @since 3.1.0 | ||
| */ | ||
| @Immutable | ||
| public record ShardedBinaryEnvelope(Object message, String entityName) {} |
107 changes: 107 additions & 0 deletions
107
...uster/src/main/java/org/eclipse/ditto/internal/utils/cluster/ShardedBinarySerializer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| /* | ||
| * Copyright (c) 2022 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.internal.utils.cluster; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.Charset; | ||
| import java.nio.charset.StandardCharsets; | ||
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| import akka.actor.ActorSystem; | ||
| import akka.actor.ExtendedActorSystem; | ||
| import akka.serialization.ByteBufferSerializer; | ||
| import akka.serialization.Serialization; | ||
| import akka.serialization.SerializationExtension; | ||
| import akka.serialization.SerializerWithStringManifest; | ||
| import akka.serialization.Serializers; | ||
|
|
||
| /** | ||
| * Serializer of {@link ShardedBinarySerializer}. | ||
| */ | ||
| public final class ShardedBinarySerializer | ||
| extends SerializerWithStringManifest implements ByteBufferSerializer { | ||
|
|
||
| private static final int UNIQUE_IDENTIFIER = 1259836351; | ||
| private static final Charset CHARSET = StandardCharsets.UTF_8; | ||
|
|
||
| private final ActorSystem actorSystem; | ||
| @Nullable private Serialization serialization; | ||
|
|
||
| /** | ||
| * Constructs a new sharded binary serializer. | ||
| * | ||
| * @param actorSystem the ExtendedActorSystem to use for serialization of wrapped messages. | ||
| */ | ||
| public ShardedBinarySerializer(final ExtendedActorSystem actorSystem) { | ||
| this.actorSystem = actorSystem; | ||
| } | ||
|
|
||
| @Override | ||
| public int identifier() { | ||
| return UNIQUE_IDENTIFIER; | ||
| } | ||
|
|
||
| @Override | ||
| public String manifest(final Object o) { | ||
| final var envelope = (ShardedBinaryEnvelope) o; | ||
| final var message = envelope.message(); | ||
| return Serializers.manifestFor(getSerialization().findSerializerFor(message), message); | ||
| } | ||
|
|
||
| @Override | ||
| public void toBinary(final Object o, final ByteBuffer buf) { | ||
| buf.put(toBinary(o)); | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] toBinary(final Object o) { | ||
| final var envelope = (ShardedBinaryEnvelope) o; | ||
| final var message = envelope.message(); | ||
| final var serialization = getSerialization(); | ||
| final int serializerId = serialization.findSerializerFor(message).identifier(); | ||
| final byte[] messageBytes = serialization.serialize(message).get(); | ||
| final byte[] entityNameBytes = envelope.entityName().getBytes(CHARSET); | ||
| final var buffer = ByteBuffer.allocate(4 + 4 + entityNameBytes.length + messageBytes.length); | ||
| buffer.putInt(serializerId); | ||
| buffer.putInt(entityNameBytes.length); | ||
| buffer.put(entityNameBytes); | ||
| buffer.put(messageBytes); | ||
| return buffer.array(); | ||
| } | ||
|
|
||
| @Override | ||
| public Object fromBinary(final byte[] bytes, final String manifest) { | ||
| return fromBinary(ByteBuffer.wrap(bytes), manifest); | ||
| } | ||
|
|
||
| @Override | ||
| public Object fromBinary(final ByteBuffer buf, final String manifest) { | ||
| final int serializerId = buf.getInt(); | ||
| final int entityNameLength = buf.getInt(); | ||
| final byte[] entityNameBytes = new byte[entityNameLength]; | ||
| buf.get(entityNameBytes); | ||
| final byte[] messageBytes = new byte[buf.remaining()]; | ||
| buf.get(messageBytes); | ||
| final var message = getSerialization().deserialize(messageBytes, serializerId, manifest).get(); | ||
| final var entityName = new String(entityNameBytes, CHARSET); | ||
| return new ShardedBinaryEnvelope(message, entityName); | ||
| } | ||
|
|
||
| private Serialization getSerialization() { | ||
| if (serialization == null) { | ||
| serialization = SerializationExtension.get(actorSystem); | ||
| } | ||
| return serialization; | ||
| } | ||
| } |
Oops, something went wrong.