2.0: Hide Pekko types from the embedding API, move the server into its module

README.md

@@ -309,15 +309,7 @@ Add ElasticMQ Rest SQS module to `build.sbt` dependencies
 libraryDependencies += "org.elasticmq" %% "elasticmq-rest-sqs" % Version
 ```
 
-Simply start the server using custom configuration (see examples above):
-
-```scala
-val config = ConfigFactory.load("elasticmq.conf")
-val server = new ElasticMQServer(new ElasticMQServerConfig(config))
-server.start()
-```
-
-Alternatively, custom rest server can be built using `SQSRestServerBuilder` provided in `elasticmq-rest-sqs` package:
+Then build and start a server with `SQSRestServerBuilder`:
 
 ```scala
 val server = SQSRestServerBuilder.start()
@@ -333,18 +325,44 @@ val server = SQSRestServerBuilder.withPort(9321).withInterface("localhost").star
 server.stopAndWait()
 ```
 
-You can also set a dynamic port with a port value of `0` or by using the method `withDynamicPort`. To retrieve the port (and other configuration) when using a dynamic port value you can access the server via `waitUntilStarted` for example:
+You can also set a dynamic port with a port value of `0` or by using the method `withDynamicPort`. To find out the
+assigned port, wait for the server to start; `waitUntilStarted` returns the bound address:
 
 ```scala
 val server = SQSRestServerBuilder.withDynamicPort().start()
-server.waitUntilStarted().localAddress().getPort()
+val port = server.waitUntilStarted().localAddress.getPort
 ```
 
-You can also provide a custom `ActorSystem`; for details see the javadocs.
+To be notified about changes in queues and messages, register a `QueueEventListener`. All its methods have empty
+defaults, so override only the ones you need:
+
+```scala
+val server = SQSRestServerBuilder
+  .withQueueEventListener(new QueueEventListener {
+    override def messageAdded(queueName: String, message: MessageData): Unit =
+      println(s"New message in $queueName: ${message.content.value}")
+  })
+  .start()
+```
 
-Embedded ElasticMQ can be used from any JVM-based language (Java, Scala, etc.).
+Callbacks run one at a time. The message callbacks run before the SQS response is sent, so keep them fast.
+
+By default the server creates its own Pekko `ActorSystem` and terminates it on `stopAndWait()`. If your application
+already runs one, pass it with `withActorSystem(actorSystem)`; the server then uses it but leaves its lifecycle to you.
+
+Embedded ElasticMQ can be used from any JVM-based language (Java, Scala, etc.):
+
+```java
+SQSRestServer server = SQSRestServerBuilder.withDynamicPort().start();
+int port = server.waitUntilStarted().localAddress().getPort();
+// ... use ...
+server.stopAndWait();
+```
 
-(Note that the embedded server created with `SQSRestServerBuilder` does not load any configuration files, so you cannot automatically create queues on startup as described above. You can of course create queues programmatically.)
+(Note that the embedded server created with `SQSRestServerBuilder` does not load any configuration files, so you cannot
+automatically create queues on startup as described above. You can of course create queues programmatically. To run a
+server driven by a configuration file, including queue persistence, use the `elasticmq-server` artifact:
+`new ElasticMQServer(new ElasticMQServerConfig(ConfigFactory.load("elasticmq.conf"))).start()`.)
 
 # Using the Amazon Java SDK to access an ElasticMQ Server
 

build.sbt

@@ -177,7 +177,7 @@ lazy val restSqs: Project = (project in file("rest/rest-sqs"))
       ) ++ common
     )
   )
-  .dependsOn(core % "compile->compile;test->test", persistenceFile, persistenceSql, commonTest % "test")
+  .dependsOn(core % "compile->compile;test->test", persistenceCore, commonTest % "test")
 
 lazy val restSqsTestingAmazonJavaSdk: Project =
   (project in file("rest/rest-sqs-testing-amazon-java-sdk"))
@@ -232,7 +232,7 @@ lazy val server: Project = (project in file("server"))
       }
     )
   )
-  .dependsOn(core, restSqs)
+  .dependsOn(core, restSqs, persistenceFile, persistenceSql, commonTest % "test")
 
 // Packages the GraalVM reachability metadata (META-INF/native-image) and native-specific config used by
 // native-server/Dockerfile. The native image is built from the Scala 3 artifacts, whose standard library

docs/BACKLOG.md

@@ -10,9 +10,9 @@ These need a major version. Ship them together in 2.0.
 
 ### Embedding API (`elasticmq-rest-sqs`)
 
-- [ ] **Make `TheSQSRestServerBuilder`'s constructor private.** It is a public case class with an 11-argument positional constructor, so every new option breaks binary compatibility. Keep only the `withX` methods and the `SQSRestServerBuilder` entry point.
-- [ ] **Hide Pekko types behind a small facade.** `withActorSystem(ActorSystem)`, `withQueueManagerActor(ActorRef)` and `withQueueEventListener(ActorRef)` expose Pekko classic types. A typed facade (e.g. a `QueueEventListener` trait) would let 2.x upgrade Pekko without breaking embedders.
-- [ ] **Move `ElasticMQServer` and `ElasticMQServerConfig` into the `server` module** (or a new `elasticmq-server-core` artifact). They are in package `org.elasticmq.server` but live in `rest-sqs`.
+- [x] **Make `TheSQSRestServerBuilder`'s constructor private.** It is a public case class with an 11-argument positional constructor, so every new option breaks binary compatibility. Keep only the `withX` methods and the `SQSRestServerBuilder` entry point.
+- [x] **Hide Pekko types behind a small facade.** `withActorSystem(ActorSystem)`, `withQueueManagerActor(ActorRef)` and `withQueueEventListener(ActorRef)` expose Pekko classic types. A typed facade (e.g. a `QueueEventListener` trait) would let 2.x upgrade Pekko without breaking embedders.
+- [x] **Move `ElasticMQServer` and `ElasticMQServerConfig` into the `server` module** (or a new `elasticmq-server-core` artifact). They are in package `org.elasticmq.server` but live in `rest-sqs`.
 
 ### Configuration
 
@@ -50,7 +50,7 @@ Additive or internal; can land in 2.0 or any 2.x minor.
 ### Configuration & docs
 
 - [ ] Move the Pekko settings from `rest-sqs/src/main/resources/application.conf` into `reference.conf`, so custom config files no longer need `include classpath("application.conf")` (existing includes keep working).
-- [ ] Fix the `SQSRestServerBuilder` Scaladoc: it says the default `sqsLimits` is relaxed, but the code uses `StrictSQSLimits`.
+- [x] Fix the `SQSRestServerBuilder` Scaladoc: it says the default `sqsLimits` is relaxed, but the code uses `StrictSQSLimits`.
 - [ ] Fix the "Amazon SQS allows 256KB message payloads" comment in `application.conf`; strict limits already allow 1 MiB. Check `max-uri-length` accordingly.
 - [ ] Restate the Docker image contract in the README: entrypoint, config path (`/opt/elasticmq.conf`), data volume (`/data`).
 - [ ] Update the README version automatically in the release flow (#1026).

docs/RELEASE-NOTES-2.0.md

@@ -0,0 +1,116 @@
+# ElasticMQ 2.0 release notes
+
+> **Draft.** 2.0 is in progress. This file covers the changes merged since 1.7.1 plus the open 2.0 work in
+> [#1275](https://github.com/softwaremill/elasticmq/pull/1275). Update it as items from [`BACKLOG.md`](BACKLOG.md) land.
+
+ElasticMQ 2.0 is a major release. It raises the platform baseline to Java 17 and Scala 2.13/3, and reworks the
+embedding API so it no longer exposes Pekko internals. If you run ElasticMQ as a stand-alone server or Docker image
+with the default configuration, you should not need to change anything. If you embed ElasticMQ in your application,
+read [Breaking changes](#breaking-changes) first.
+
+## Requirements
+
+- **Java 17 or newer** (1.x ran on Java 8).
+- **Scala 3 or 2.13.** Scala 2.12 artifacts are no longer published. Scala 3 is now the default build.
+
+## Breaking changes
+
+### Embedding API (`elasticmq-rest-sqs`)
+
+#### The builder is configured only through `withX` methods
+
+`TheSQSRestServerBuilder` is no longer a case class. Its constructor, fields and `copy` are private. Start from
+`SQSRestServerBuilder` and chain the `withX` methods:
+
+```scala
+// 1.x
+val server = TheSQSRestServerBuilder(None, None, "", 9324, NodeAddress(), true, StrictSQSLimits, "elasticmq",
+  "000000000000", None).start()
+
+// 2.0
+val server = SQSRestServerBuilder.withPort(9324).withSQSLimits(StrictSQSLimits).start()
+```
+
+This lets future 2.x releases add builder options without breaking binary compatibility.
+
+#### `withQueueEventListener` takes a `QueueEventListener` instead of an `ActorRef`
+
+In 1.x the listener was a Pekko actor that had to understand ElasticMQ's internal `QueueEvent` messages and reply to
+some of them. In 2.0 it is a plain trait with optional callbacks, usable from Scala and Java:
+
+```scala
+// 2.0
+SQSRestServerBuilder
+  .withQueueEventListener(new QueueEventListener {
+    override def queueCreated(queue: QueueData): Unit = ???
+    override def messageAdded(queueName: String, message: MessageData): Unit = ???
+  })
+  .start()
+```
+
+Available callbacks: `queueCreated`, `queueDeleted`, `queueMetadataUpdated`, `messageAdded`, `messageUpdated`,
+`messageRemoved`. They are called one at a time, in the order the changes happen. The message callbacks finish
+before the SQS response is sent, so keep them fast; the server waits for them for up to 5 seconds. An exception thrown
+by a callback is logged and doesn't fail the request.
+
+#### `withQueueManagerActor` is no longer public
+
+The builder always creates its own queue manager. Wiring a file or SQL persistence actor into an embedded server is no
+longer supported. To use persistence, run the stand-alone server (`elasticmq-server`) or open an issue describing
+your use case.
+
+#### `SQSRestServer` no longer exposes Pekko HTTP types
+
+- `waitUntilStarted()` returns an `SQSRestServerBinding` instead of Pekko's `Http.ServerBinding`. The bound address
+  is still available as `localAddress` (a `java.net.InetSocketAddress`):
+
+  ```scala
+  val port = SQSRestServerBuilder.withDynamicPort().start().waitUntilStarted().localAddress.getPort
+  ```
+
+- `SQSRestServer` is no longer a case class and can't be constructed directly.
+- `stopAndGetFuture()` returns `Future[Unit]` instead of `Future[Any]`.
+
+#### Still supported: your own actor system
+
+`withActorSystem(ActorSystem)` is unchanged. The server runs its actors and HTTP layer on the given system and leaves
+stopping it to you.
+
+### `ElasticMQServer` moved to `elasticmq-server`
+
+`ElasticMQServer` and `ElasticMQServerConfig` (package `org.elasticmq.server`) moved from `elasticmq-rest-sqs` to
+`elasticmq-server`, together with the server's default configuration (`reference.conf`). Package names are unchanged.
+If you start a configuration-driven server from code, change the dependency:
+
+```scala
+libraryDependencies += "org.elasticmq" %% "elasticmq-server" % "2.0.0"
+```
+
+As a result, `elasticmq-rest-sqs` no longer depends on `elasticmq-persistence-file` and `elasticmq-persistence-sql`, so
+embedding it no longer pulls in H2, ScalikeJDBC and PureConfig.
+
+### Native image
+
+The native image is built with GraalVM for JDK 25 and is now a mostly static binary running on a distroless base
+image (`gcr.io/distroless/base-nossl-debian13`) instead of Alpine. The image no longer contains a shell or `tini`, so
+derived images that run shell commands need to change. This also fixes DNS resolution, such as binding to a Docker
+network alias.
+
+## Bug fixes
+
+- `ReceiveMessage` matches requested message attribute names exactly or by prefix (`prefix.*`), as SQS does
+  (#1268).
+- Queue events now carry a snapshot of the message. Persistence could fail with `ConcurrentModificationException`
+  while the queue kept updating a message (#1270).
+
+## UI
+
+- Works with real AWS accounts: when `SQS_ENDPOINT` is unset, the SDK uses the default AWS endpoint.
+- New `NEXT_PUBLIC_AUTO_REFRESH_DISABLED` flag turns off the 1-second polling, which is recommended against real AWS
+  to avoid API costs.
+- Runs on Node.js 26 LTS.
+
+## Dependencies
+
+- Scala 3.9 LTS, Pekko 1.7.1, Pekko HTTP 1.4.2, H2 2.5.252, Logback 1.6.3.
+- Build moved to sbt 2.

performance-tests/src/test/scala/org/elasticmq/performance/LocalPerformanceTest.scala

@@ -97,7 +97,7 @@ object LocalPerformanceTest extends App {
   trait MQWithQueueManagerActor extends MQ {
     private var currentQueue: ActorRef = _
 
-    implicit val timeout = Timeout(10000L, TimeUnit.MILLISECONDS)
+    implicit val timeout: Timeout = Timeout(10000L, TimeUnit.MILLISECONDS)
 
     override def start(sqsLimits: Limits) = {
       val queueManagerActor = super.start(sqsLimits)
@@ -170,7 +170,7 @@ object LocalPerformanceTest extends App {
 
       currentRestServer = SQSRestServerBuilder
         .withActorSystem(actorSystem)
-        .withQueueManagerActor(queueManagerActor)
+        .withSQSLimits(sqsLimits)
         .start()
 
       currentSQSClient = AmazonSQSClientBuilder
@@ -186,7 +186,7 @@ object LocalPerformanceTest extends App {
     }
 
     override def stop(): Unit = {
-      currentRestServer.stopAndGetFuture
+      currentRestServer.stopAndWait()
       currentSQSClient.shutdown()
       super.stop()
     }

rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/integration/QueueEventListenerTests.scala

@@ -0,0 +1,101 @@
+package org.elasticmq.rest.sqs.integration
+
+import org.elasticmq.rest.sqs.{QueueEventListener, SQSRestServerBuilder}
+import org.elasticmq.{MessageData, QueueData}
+import org.scalatest.concurrent.Eventually
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.sqs.model._
+import software.amazon.awssdk.services.sqs.{SqsClient => AwsSqsClient}
+
+import java.net.URI
+import java.util.concurrent.ConcurrentLinkedQueue
+import scala.jdk.CollectionConverters._
+import scala.util.Try
+
+class QueueEventListenerTests extends AnyFunSuite with Matchers with Eventually {
+
+  private class RecordingListener extends QueueEventListener {
+    val events = new ConcurrentLinkedQueue[String]()
+
+    override def queueCreated(queue: QueueData): Unit = events.add(s"queueCreated:${queue.name}")
+    override def queueDeleted(queueName: String): Unit = events.add(s"queueDeleted:$queueName")
+    override def queueMetadataUpdated(queue: QueueData): Unit =
+      events.add(s"queueMetadataUpdated:${queue.name}:${queue.defaultVisibilityTimeout.seconds}")
+
+    override def messageAdded(queueName: String, message: MessageData): Unit =
+      events.add(s"messageAdded:$queueName:${message.content.value}")
+    override def messageUpdated(queueName: String, message: MessageData): Unit =
+      events.add(s"messageUpdated:$queueName:${message.content.value}:${message.statistics.approximateReceiveCount}")
+    override def messageRemoved(queueName: String, messageId: String): Unit =
+      events.add(s"messageRemoved:$queueName")
+
+    def recorded: List[String] = events.asScala.toList
+  }
+
+  private def withServer(listener: QueueEventListener)(test: AwsSqsClient => Unit): Unit = {
+    val server = SQSRestServerBuilder.withDynamicPort().withQueueEventListener(listener).start()
+    val port = server.waitUntilStarted().localAddress.getPort
+
+    val client = AwsSqsClient
+      .builder()
+      .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
+      .region(Region.EU_CENTRAL_1)
+      .endpointOverride(new URI(s"http://localhost:$port"))
+      .build()
+
+    try test(client)
+    finally {
+      Try(client.close())
+      Try(server.stopAndWait())
+    }
+  }
+
+  test("should notify the listener about queue and message changes") {
+    val listener = new RecordingListener
+    withServer(listener) { client =>
+      val queueUrl = client.createQueue(CreateQueueRequest.builder().queueName("q1").build()).queueUrl()
+      eventually(listener.recorded shouldBe List("queueCreated:q1"))
+
+      client.setQueueAttributes(
+        SetQueueAttributesRequest
+          .builder()
+          .queueUrl(queueUrl)
+          .attributes(Map(QueueAttributeName.VISIBILITY_TIMEOUT -> "15").asJava)
+          .build()
+      )
+      eventually(listener.recorded should contain("queueMetadataUpdated:q1:15"))
+
+      // message callbacks complete before the SQS response is sent, so no need to wait for them
+      client.sendMessage(SendMessageRequest.builder().queueUrl(queueUrl).messageBody("hello").build())
+      listener.recorded.last shouldBe "messageAdded:q1:hello"
+
+      val received = client.receiveMessage(ReceiveMessageRequest.builder().queueUrl(queueUrl).build()).messages()
+      listener.recorded.last shouldBe "messageUpdated:q1:hello:1"
+
+      client.deleteMessage(
+        DeleteMessageRequest.builder().queueUrl(queueUrl).receiptHandle(received.get(0).receiptHandle()).build()
+      )
+      listener.recorded.last shouldBe "messageRemoved:q1"
+
+      client.deleteQueue(DeleteQueueRequest.builder().queueUrl(queueUrl).build())
+      eventually(listener.recorded.last shouldBe "queueDeleted:q1")
+    }
+  }
+
+  test("should keep serving requests when the listener throws") {
+    val failingListener = new QueueEventListener {
+      override def queueCreated(queue: QueueData): Unit = throw new RuntimeException("boom")
+      override def messageAdded(queueName: String, message: MessageData): Unit = throw new RuntimeException("boom")
+    }
+    withServer(failingListener) { client =>
+      val queueUrl = client.createQueue(CreateQueueRequest.builder().queueName("q2").build()).queueUrl()
+      client.sendMessage(SendMessageRequest.builder().queueUrl(queueUrl).messageBody("hello").build())
+
+      val received = client.receiveMessage(ReceiveMessageRequest.builder().queueUrl(queueUrl).build()).messages()
+      received.asScala.map(_.body()) shouldBe List("hello")
+    }
+  }
+}