diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala index fd847b3fe2..4e051d5732 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala @@ -131,13 +131,15 @@ private[pekko] final class ArterySettings private (config: Config) { s"tcp-magic value [$first] must produce at least 4 UTF-8 bytes, but produced [${bytes.length}] bytes") bytes.take(4) } - val TcpMagicValues: Set[ByteString] = { + // A Seq rather than a Set: a Set hashes its members, and hashing a ByteString walks all its + // bytes, where comparing this handful of 4-byte values is a couple of cheap equality checks. + val TcpMagicValues: immutable.Seq[ByteString] = { tcpMagicList.map { s => val bytes = ByteString(s.getBytes(StandardCharsets.UTF_8)) require(bytes.length >= 4, s"tcp-magic value [$s] must produce at least 4 UTF-8 bytes, but produced [${bytes.length}] bytes") bytes.take(4) - }.toSet + }.distinct.toList } val Dispatcher: String = getString("use-dispatcher") val ControlStreamDispatcher: String = getString("use-control-stream-dispatcher") diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala index 39da7b75c6..d752ecd015 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala @@ -17,6 +17,8 @@ package tcp import java.nio.ByteBuffer import java.nio.ByteOrder +import scala.collection.immutable + import org.apache.pekko import pekko.annotation.InternalApi import pekko.stream.Attributes @@ -76,7 +78,7 @@ import pekko.util.ByteString * INTERNAL API */ @InternalApi private[pekko] class TcpFraming( - acceptedMagic: Set[ByteString] = Set(TcpFraming.DefaultMagic), + acceptedMagic: immutable.Seq[ByteString] = List(TcpFraming.DefaultMagic), flightRecorder: RemotingFlightRecorder = NoOpRemotingFlightRecorder) extends ByteStringParser[EnvelopeBuffer] { diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/ArterySettingsSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/ArterySettingsSpec.scala new file mode 100644 index 0000000000..23f2344c91 --- /dev/null +++ b/remote/src/test/scala/org/apache/pekko/remote/artery/ArterySettingsSpec.scala @@ -0,0 +1,81 @@ +/* + * 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.remote.artery + +import java.nio.charset.StandardCharsets + +import org.apache.pekko +import pekko.util.ByteString + +import com.typesafe.config.ConfigFactory +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class ArterySettingsSpec extends AnyWordSpec with Matchers { + + private def settings(tcpMagic: String): ArterySettings = + ArterySettings( + ConfigFactory + .parseString(s"pekko.remote.artery.advanced.tcp-magic = $tcpMagic") + .withFallback(ConfigFactory.load()) + .resolve() + .getConfig("pekko.remote.artery")) + + private def magic(s: String): ByteString = ByteString(s.getBytes(StandardCharsets.UTF_8)) + + "ArterySettings.TcpMagicValues" must { + + // Held as a Seq rather than a Set: a Set hashes its members, and hashing a ByteString walks + // all of its bytes. These assertions pin the ordering and de-duplication that the Set + // previously provided incidentally, so the collection type cannot be changed back silently. + "default to AKKA then AKKA, in configuration order" in { + val defaults = ArterySettings(ConfigFactory.load().getConfig("pekko.remote.artery")) + defaults.Advanced.TcpMagicValues should ===(List(magic("AKKA"), magic("PEKK"))) + defaults.Advanced.TcpMagic should ===(magic("AKKA")) + } + + "preserve configuration order" in { + settings("""["AKKA", "PEKK"]""").Advanced.TcpMagicValues should ===(List(magic("AKKA"), magic("PEKK"))) + } + + "use the first configured value as the outbound magic" in { + settings("""["AKKA", "PEKK"]""").Advanced.TcpMagic should ===(magic("AKKA")) + } + + "drop duplicates, keeping the first occurrence" in { + settings("""["PEKK", "AKKA", "PEKK"]""").Advanced.TcpMagicValues should ===( + List(magic("PEKK"), magic("AKKA"))) + } + + "truncate each value to 4 bytes, and de-duplicate after truncating" in { + settings("""["PEKKO"]""").Advanced.TcpMagicValues should ===(List(magic("PEKK"))) + settings("""["PEKKO", "PEKK"]""").Advanced.TcpMagicValues should ===(List(magic("PEKK"))) + } + + // Advanced is an object, so it is initialised lazily: the requires do not run until one of + // its members is touched, which is why these force TcpMagicValues rather than just building + // the settings. + "reject a value shorter than 4 UTF-8 bytes" in { + an[IllegalArgumentException] should be thrownBy settings("""["PEK"]""").Advanced.TcpMagicValues + } + + "reject an empty list" in { + an[IllegalArgumentException] should be thrownBy settings("[]").Advanced.TcpMagicValues + } + } +} diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala index 39f5e2ee18..f27d0547d8 100644 --- a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala +++ b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala @@ -31,7 +31,7 @@ class TcpFramingSpec extends PekkoSpec(""" import TcpFraming.encodeFrameHeader private val magic = TcpFraming.DefaultMagic - private val acceptedMagic = Set(magic, TcpFraming.PekkoMagic) + private val acceptedMagic = List(magic, TcpFraming.PekkoMagic) private val framingFlow = Flow[ByteString].via(new TcpFraming(acceptedMagic)) private val payload5 = ByteString((1 to 5).map(_.toByte).toArray) @@ -117,7 +117,7 @@ class TcpFramingSpec extends PekkoSpec(""" "accept custom magic" in { val customMagic = ByteString('T'.toByte, 'E'.toByte, 'S'.toByte, 'T'.toByte) - val customFramingFlow = Flow[ByteString].via(new TcpFraming(Set(customMagic))) + val customFramingFlow = Flow[ByteString].via(new TcpFraming(List(customMagic))) val bytes = TcpFraming.encodeConnectionHeader(customMagic, 2) ++ frameBytes(1) val frames = Source(List(bytes)).via(customFramingFlow).runWith(Sink.seq).futureValue frames.head.streamId should ===(2)