This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 718bc072bb filter messages from remember-entities store (#3411)
718bc072bb is described below
commit 718bc072bbaed2c37254a0372521ddb506c59916
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Aug 7 12:57:29 2026 +0100
filter messages from remember-entities store (#3411)
* filter messages from remember-entities store
* Update ExtractorAdapterSpec.scala
* Update ExtractorAdapterSpec.scala
* Update ExtractorAdapterSpec.scala
* Update ExtractorAdapterSpec.scala
---
.../typed/internal/ClusterShardingImpl.scala | 15 +++-
.../typed/internal/ExtractorAdapterSpec.scala | 90 ++++++++++++++++++++++
2 files changed, 101 insertions(+), 4 deletions(-)
diff --git
a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala
b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala
index 9031d7aaf3..eb2968e6f2 100644
---
a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala
+++
b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala
@@ -45,6 +45,7 @@ import pekko.cluster.sharding.ShardCoordinator
import pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
import pekko.cluster.sharding.ShardRegion
import pekko.cluster.sharding.ShardRegion.{ StartEntity => ClassicStartEntity }
+import pekko.cluster.sharding.internal.RememberEntitiesShardStore
import pekko.cluster.sharding.typed.scaladsl.EntityContext
import pekko.cluster.typed.Cluster
import pekko.event.Logging
@@ -59,14 +60,18 @@ import pekko.util.Timeout
* INTERNAL API
* Extracts entityId and unwraps ShardingEnvelope and StartEntity messages.
* Other messages are delegated to the given `ShardingMessageExtractor`.
+ * Internal messages from the remember-entities store are filtered out
+ * and should never reach the user-supplied extractor.
*/
@InternalApi private[pekko] class ExtractorAdapter[E, M](delegate:
ShardingMessageExtractor[E, M])
extends ShardingMessageExtractor[Any, M] {
override def entityId(message: Any): String = {
message match {
- case ShardingEnvelope(entityId, _) => entityId // also covers
ClassicStartEntity in ShardingEnvelope
- case ClassicStartEntity(entityId) => entityId
- case msg =>
delegate.entityId(msg.asInstanceOf[E])
+ case ShardingEnvelope(entityId, _) => entityId //
also covers ClassicStartEntity in ShardingEnvelope
+ case ClassicStartEntity(entityId) => entityId
+ case _: RememberEntitiesShardStore.UpdateDone => null
+ case _: RememberEntitiesShardStore.RememberedEntities => null
+ case msg =>
delegate.entityId(msg.asInstanceOf[E])
}
}
@@ -80,7 +85,9 @@ import pekko.util.Timeout
case msg: ClassicStartEntity =>
// not really of type M, but erased and StartEntity is only handled
internally, not delivered to the entity
msg.asInstanceOf[M]
- case msg =>
+ case _: RememberEntitiesShardStore.UpdateDone =>
null.asInstanceOf[M]
+ case _: RememberEntitiesShardStore.RememberedEntities =>
null.asInstanceOf[M]
+ case msg =>
delegate.unwrapMessage(msg.asInstanceOf[E])
}
}
diff --git
a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ExtractorAdapterSpec.scala
b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ExtractorAdapterSpec.scala
new file mode 100644
index 0000000000..b19bb111a0
--- /dev/null
+++
b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ExtractorAdapterSpec.scala
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import scala.annotation.nowarn
+
+import org.apache.pekko
+import pekko.cluster.sharding.ShardRegion.{ StartEntity => ClassicStartEntity }
+import pekko.cluster.sharding.internal.RememberEntitiesShardStore
+import pekko.cluster.sharding.typed.ShardingEnvelope
+import pekko.cluster.sharding.typed.ShardingMessageExtractor
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+@nowarn("cat=lint-structural-type")
+class ExtractorAdapterSpec extends AnyWordSpecLike with Matchers {
+
+ private val extractor = new ShardingMessageExtractor[String, String] {
+ override def entityId(message: String): String =
+ if (message.startsWith("entity-")) message.substring(7,
message.indexOf(':'))
+ else null
+
+ override def shardId(entityId: String): String =
entityId.hashCode.abs.toString
+
+ override def unwrapMessage(message: String): String =
message.substring(message.indexOf(':') + 1)
+ }
+
+ private val adapter = new ExtractorAdapter(extractor)
+
+ "ExtractorAdapter" must {
+
+ "extract entity id from ShardingEnvelope" in {
+ adapter.entityId(ShardingEnvelope("entity-1", "hello")) should
===("entity-1")
+ }
+
+ "extract entity id from ClassicStartEntity" in {
+ adapter.entityId(ClassicStartEntity("entity-1")) should ===("entity-1")
+ }
+
+ "delegate to user extractor for user messages" in {
+ adapter.entityId("entity-1:hello") should ===("1")
+ }
+
+ "return null for RememberEntitiesShardStore.UpdateDone" in {
+ adapter.entityId(RememberEntitiesShardStore.UpdateDone(Set("entity-1"),
Set.empty)) should be(null)
+ }
+
+ "return null for RememberEntitiesShardStore.RememberedEntities" in {
+
adapter.entityId(RememberEntitiesShardStore.RememberedEntities(Set("entity-1",
"entity-2"))) should be(null)
+ }
+
+ "unwrap ShardingEnvelope message" in {
+ assert(adapter.unwrapMessage(ShardingEnvelope("entity-1", "hello")) ===
"hello")
+ }
+
+ "unwrap ClassicStartEntity message" in {
+ val msg = ClassicStartEntity("entity-1")
+ // widen to Any to avoid checkcast to M at the call site
(ClassicStartEntity is not actually M)
+ assert((adapter.unwrapMessage(msg): Any) === msg)
+ }
+
+ "delegate unwrapMessage to user extractor for user messages" in {
+ assert(adapter.unwrapMessage("entity-1:hello") === "hello")
+ }
+
+ "return null for unwrapMessage on UpdateDone" in {
+
assert(adapter.unwrapMessage(RememberEntitiesShardStore.UpdateDone(Set("entity-1"),
Set.empty)) === null)
+ }
+
+ "return null for unwrapMessage on RememberedEntities" in {
+
assert(adapter.unwrapMessage(RememberEntitiesShardStore.RememberedEntities(Set("entity-1")))
=== null)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
