diff --git a/build.sbt b/build.sbt
index 5e12dd8..6d9cd3d 100644
--- a/build.sbt
+++ b/build.sbt
@@ -61,6 +61,7 @@ lazy val `pekko-http-json` =
       `pekko-http-avro4s`,
       `pekko-http-circe`,
       `pekko-http-jackson`,
+      `pekko-http-jackson3`,
       `pekko-http-json4s`,
       `pekko-http-jsoniter-scala`,
       `pekko-http-ninny`,
@@ -107,10 +108,22 @@ lazy val `pekko-http-jackson` =
     .settings(
       libraryDependencies ++= Seq(
         library.pekkoHttp,
-        library.jacksonModuleScala,
-        library.pekkoStream             % Provided,
-        library.scalaTest               % Test,
-        library.jacksonModuleParamNames % Test
+        library.jacksonModuleScala2,
+        library.pekkoStream              % Provided,
+        library.scalaTest                % Test,
+        library.jacksonModuleParamNames2 % Test
+      )
+    )
+
+lazy val `pekko-http-jackson3` =
+  project
+    .settings(commonSettings, withScala3)
+    .settings(
+      libraryDependencies ++= Seq(
+        library.pekkoHttp,
+        library.jacksonModuleScala3,
+        library.pekkoStream % Provided,
+        library.scalaTest   % Test
       )
     )
 
@@ -221,19 +234,20 @@ lazy val commonSettings =
 lazy val library =
   new {
     object Version {
-      val pekko              = "1.1.3"
-      val pekkoHttp          = "1.1.0"
-      val argonaut           = "6.3.11"
-      val avro4s             = "4.1.2"
-      val circe              = "0.14.10"
-      val jacksonModuleScala = "2.18.3"
-      val json4s             = "4.0.7"
-      val jsoniterScala      = "2.33.2"
-      val ninny              = "0.9.1"
-      val play               = "3.0.4"
-      val scalaTest          = "3.2.19"
-      val upickle            = "4.1.0"
-      val zioJson            = "0.7.36"
+      val pekko         = "1.1.3"
+      val pekkoHttp     = "1.1.0"
+      val argonaut      = "6.3.11"
+      val avro4s        = "4.1.2"
+      val circe         = "0.14.10"
+      val jackson2      = "2.18.3"
+      val jackson3      = "3.0.0-rc1"
+      val json4s        = "4.0.7"
+      val jsoniterScala = "2.33.2"
+      val ninny         = "0.9.1"
+      val play          = "3.0.4"
+      val scalaTest     = "3.2.19"
+      val upickle       = "4.1.0"
+      val zioJson       = "0.7.36"
     }
     // format: off
     val pekkoHttp            = "org.apache.pekko"                      %% "pekko-http"            % Version.pekkoHttp
@@ -243,8 +257,9 @@ lazy val library =
     val circe                = "io.circe"                              %% "circe-core"            % Version.circe
     val circeGeneric         = "io.circe"                              %% "circe-generic"         % Version.circe
     val circeParser          = "io.circe"                              %% "circe-parser"          % Version.circe
-    val jacksonModuleScala   = "com.fasterxml.jackson.module"          %% "jackson-module-scala"  % Version.jacksonModuleScala
-    val jacksonModuleParamNames = "com.fasterxml.jackson.module"        % "jackson-module-parameter-names" % Version.jacksonModuleScala
+    val jacksonModuleScala2  = "com.fasterxml.jackson.module"          %% "jackson-module-scala"  % Version.jackson2
+    val jacksonModuleParamNames2 = "com.fasterxml.jackson.module"       % "jackson-module-parameter-names" % Version.jackson2
+    val jacksonModuleScala3  = "tools.jackson.module"                  %% "jackson-module-scala"  % Version.jackson3
     val json4sCore           = "org.json4s"                            %% "json4s-core"           % Version.json4s
     val json4sJackson        = "org.json4s"                            %% "json4s-jackson"        % Version.json4s
     val json4sNative         = "org.json4s"                            %% "json4s-native"         % Version.json4s
diff --git a/pekko-http-jackson3/src/main/java/com/github/pjfanning/pekkohttpjackson3/Jackson.java b/pekko-http-jackson3/src/main/java/com/github/pjfanning/pekkohttpjackson3/Jackson.java
new file mode 100644
index 0000000..e36a81c
--- /dev/null
+++ b/pekko-http-jackson3/src/main/java/com/github/pjfanning/pekkohttpjackson3/Jackson.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package com.github.pjfanning.pekkohttpjackson3;
+
+import org.apache.pekko.http.javadsl.model.MediaTypes;
+import org.apache.pekko.http.javadsl.model.RequestEntity;
+import org.apache.pekko.http.javadsl.marshalling.Marshaller;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.databind.ObjectMapper;
+
+final class Jackson {
+
+    static <T> Marshaller<T, RequestEntity> marshaller(ObjectMapper mapper) {
+        return Marshaller.wrapEntity(
+                u -> toJSON(mapper, u), Marshaller.stringToEntity(), MediaTypes.APPLICATION_JSON);
+    }
+
+    private static String toJSON(ObjectMapper mapper, Object object) {
+        try {
+            return mapper.writeValueAsString(object);
+        } catch (JacksonException e) {
+            throw new IllegalArgumentException("Cannot marshal to JSON: " + object, e);
+        }
+    }
+
+    private Jackson() {}
+}
diff --git a/pekko-http-jackson3/src/main/resources/reference.conf b/pekko-http-jackson3/src/main/resources/reference.conf
new file mode 100644
index 0000000..19b650a
--- /dev/null
+++ b/pekko-http-jackson3/src/main/resources/reference.conf
@@ -0,0 +1,41 @@
+pekko-http-json {
+  jackson {
+    jackson-modules += "tools.jackson.module.scala.DefaultScalaModule"
+    read {
+      # see https://www.javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.18.0/com/fasterxml/jackson/core/StreamReadConstraints.html
+      # these defaults are the same as the defaults in `StreamReadConstraints`
+      max-nesting-depth = 1000
+      max-number-length = 1000
+      max-string-length = 20000000
+      # added in jackson 2.16.0
+      max-name-length = 50000
+      # max-document-length of -1 means unlimited (since jackson 2.16.0)
+      max-document-length = -1
+      # max-token-count of -1 means unlimited (since jackson 2.18.0)
+      max-token-count = -1
+
+      # see https://www.javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.18.0/com/fasterxml/jackson/core/StreamReadFeature.html
+      # these defaults are the same as the defaults in `StreamReadFeature`
+      feature {
+        include-source-in-location = false
+      }
+    }
+    write {
+      # see https://www.javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.18.0/com/fasterxml/jackson/core/StreamWriteConstraints.html
+      # these defaults are the same as the defaults in `StreamWriteConstraints`
+      max-nesting-depth = 1000
+    }
+
+    # Controls the Buffer Recycler Pool implementation used by Jackson.
+    # https://javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.18.0/com/fasterxml/jackson/core/util/JsonRecyclerPools.html
+    # The default is "thread-local" which is the same as the default in Jackson 2.16.
+    buffer-recycler {
+      # the supported values are "thread-local", "concurrent-deque", "shared-concurrent-deque", "bounded", "non-recycling"
+      # "lock-free", "shared-lock-free" are supported but not recommended as they are due for removal in Jackson
+      pool-instance = "thread-local"
+      # the maximum size of bounded recycler pools - must be >=1 or an IllegalArgumentException will occur
+      # only applies to pool-instance type "bounded"
+      bounded-pool-size = 100
+    }
+  }
+}
diff --git a/pekko-http-jackson3/src/main/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupport.scala b/pekko-http-jackson3/src/main/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupport.scala
new file mode 100644
index 0000000..add3d20
--- /dev/null
+++ b/pekko-http-jackson3/src/main/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupport.scala
@@ -0,0 +1,263 @@
+/*
+ * Copyright 2015 Heiko Seeberger
+ *
+ * Licensed 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 com.github.pjfanning.pekkohttpjackson3
+
+import tools.jackson.core.util.{ BufferRecycler, JsonRecyclerPools, RecyclerPool }
+import tools.jackson.core.{
+  JsonParser,
+  ObjectReadContext,
+  StreamReadConstraints,
+  StreamReadFeature,
+  StreamWriteConstraints
+}
+import tools.jackson.core.json.{ JsonFactory, JsonFactoryBuilder }
+import tools.jackson.core.async.ByteBufferFeeder
+import tools.jackson.databind.{ DeserializationFeature, JacksonModule, ObjectMapper }
+import tools.jackson.databind.json.JsonMapper
+import tools.jackson.module.scala.{ ClassTagExtensions, JavaTypeable }
+import com.typesafe.config.{ Config, ConfigFactory }
+import org.apache.pekko.http.javadsl.common.JsonEntityStreamingSupport
+import org.apache.pekko.http.scaladsl.common.EntityStreamingSupport
+import org.apache.pekko.http.scaladsl.marshalling._
+import org.apache.pekko.http.scaladsl.model.{
+  ContentTypeRange,
+  HttpEntity,
+  MediaType,
+  MessageEntity
+}
+import org.apache.pekko.http.scaladsl.model.MediaTypes.`application/json`
+import org.apache.pekko.http.scaladsl.unmarshalling.{
+  FromEntityUnmarshaller,
+  Unmarshal,
+  Unmarshaller
+}
+import org.apache.pekko.http.scaladsl.util.FastFuture
+import org.apache.pekko.stream.scaladsl.{ Flow, Source }
+import org.apache.pekko.util.ByteString
+
+import scala.collection.immutable.Seq
+import scala.concurrent.Future
+import scala.util.Try
+import scala.util.control.NonFatal
+
+/**
+  * Automatic to and from JSON marshalling/unmarshalling using an in-scope Jackson's ObjectMapper
+  */
+object JacksonSupport extends JacksonSupport {
+
+  private[pekkohttpjackson3] val jacksonConfig =
+    ConfigFactory.load().getConfig("pekko-http-json.jackson")
+
+  private[pekkohttpjackson3] def createJsonFactory(config: Config): JsonFactory = {
+    val streamReadConstraints = StreamReadConstraints
+      .builder()
+      .maxNestingDepth(config.getInt("read.max-nesting-depth"))
+      .maxNumberLength(config.getInt("read.max-number-length"))
+      .maxStringLength(config.getInt("read.max-string-length"))
+      .maxNameLength(config.getInt("read.max-name-length"))
+      .maxDocumentLength(config.getInt("read.max-document-length"))
+      .maxTokenCount(config.getInt("read.max-token-count"))
+      .build()
+    val streamWriteConstraints = StreamWriteConstraints
+      .builder()
+      .maxNestingDepth(config.getInt("write.max-nesting-depth"))
+      .build()
+    val jsonFactoryBuilder: JsonFactoryBuilder = JsonFactory
+      .builder()
+      .asInstanceOf[JsonFactoryBuilder]
+      .streamReadConstraints(streamReadConstraints)
+      .streamWriteConstraints(streamWriteConstraints)
+      .recyclerPool(getBufferRecyclerPool(config))
+      .configure(
+        StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION,
+        config.getBoolean("read.feature.include-source-in-location")
+      )
+    jsonFactoryBuilder.build()
+  }
+
+  private def getBufferRecyclerPool(cfg: Config): RecyclerPool[BufferRecycler] =
+    cfg.getString("buffer-recycler.pool-instance") match {
+      case "thread-local"            => JsonRecyclerPools.threadLocalPool()
+      case "concurrent-deque"        => JsonRecyclerPools.newConcurrentDequePool()
+      case "shared-concurrent-deque" => JsonRecyclerPools.sharedConcurrentDequePool()
+      case "bounded" =>
+        JsonRecyclerPools.newBoundedPool(cfg.getInt("buffer-recycler.bounded-pool-size"))
+      case "non-recycling" => JsonRecyclerPools.nonRecyclingPool()
+      case other           => throw new IllegalArgumentException(s"Unknown recycler-pool: $other")
+    }
+
+  val defaultObjectMapper: ObjectMapper with ClassTagExtensions = createObjectMapper(jacksonConfig)
+
+  private[pekkohttpjackson3] def createObjectMapper(
+      config: Config
+  ): ObjectMapper with ClassTagExtensions = {
+    val builder = JsonMapper.builder(createJsonFactory(config))
+    builder.disable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+    import org.apache.pekko.util.ccompat.JavaConverters._
+    val configuredModules = config.getStringList("jackson-modules").asScala.toSeq
+    val modules           = configuredModules.map(loadModule)
+    modules.foreach(builder.addModule)
+    builder.build() :: ClassTagExtensions
+  }
+
+  private def loadModule(fcqn: String): JacksonModule = {
+    val inst = Try(Class.forName(fcqn).getConstructor().newInstance())
+      .getOrElse(Class.forName(fcqn + "$").getConstructor().newInstance())
+    inst.asInstanceOf[JacksonModule]
+  }
+}
+
+/**
+  * JSON marshalling/unmarshalling using an in-scope Jackson's ObjectMapper
+  */
+trait JacksonSupport {
+  type SourceOf[A] = Source[A, _]
+
+  import JacksonSupport._
+
+  def unmarshallerContentTypes: Seq[ContentTypeRange] =
+    mediaTypes.map(ContentTypeRange.apply)
+
+  private val defaultMediaTypes: Seq[MediaType.WithFixedCharset] = List(`application/json`)
+  def mediaTypes: Seq[MediaType.WithFixedCharset]                = defaultMediaTypes
+
+  private val jsonStringUnmarshaller =
+    Unmarshaller.byteStringUnmarshaller
+      .forContentTypes(unmarshallerContentTypes: _*)
+      .mapWithCharset {
+        case (ByteString.empty, _) => throw Unmarshaller.NoContentException
+        case (data, charset)       => data.decodeString(charset.nioCharset.name)
+      }
+
+  private def sourceByteStringMarshaller(
+      mediaType: MediaType.WithFixedCharset
+  ): Marshaller[SourceOf[ByteString], MessageEntity] =
+    Marshaller[SourceOf[ByteString], MessageEntity] { implicit ec => value =>
+      try
+        FastFuture.successful {
+          Marshalling.WithFixedContentType(
+            mediaType,
+            () => HttpEntity(contentType = mediaType, data = value)
+          ) :: Nil
+        }
+      catch {
+        case NonFatal(e) => FastFuture.failed(e)
+      }
+    }
+
+  private val jsonSourceStringMarshaller =
+    Marshaller.oneOf(mediaTypes: _*)(sourceByteStringMarshaller)
+
+  private def jsonSource[A](entitySource: SourceOf[A])(implicit
+      objectMapper: ObjectMapper = defaultObjectMapper,
+      support: JsonEntityStreamingSupport
+  ): SourceOf[ByteString] =
+    entitySource
+      .map(objectMapper.writeValueAsBytes)
+      .map(ByteString(_))
+      .via(support.framingRenderer)
+
+  /**
+    * HTTP entity => `A`
+    */
+  implicit def unmarshaller[A: JavaTypeable](implicit
+      objectMapper: ObjectMapper with ClassTagExtensions = defaultObjectMapper
+  ): FromEntityUnmarshaller[A] =
+    jsonStringUnmarshaller.map(data => objectMapper.readValue[A](data))
+
+  /**
+    * `A` => HTTP entity
+    */
+  implicit def marshaller[Object](implicit
+      objectMapper: ObjectMapper = defaultObjectMapper
+  ): ToEntityMarshaller[Object] =
+    Jackson.marshaller[Object](objectMapper)
+
+  /**
+    * `ByteString` => `A`
+    *
+    * @tparam A
+    *   type to decode
+    * @return
+    *   unmarshaller for any `A` value
+    */
+  implicit def fromByteStringUnmarshaller[A: JavaTypeable](implicit
+      objectMapper: ObjectMapper with ClassTagExtensions = defaultObjectMapper
+  ): Unmarshaller[ByteString, A] =
+    Unmarshaller { ec => bs =>
+      Future {
+        val parser = objectMapper
+          .tokenStreamFactory()
+          .createNonBlockingByteBufferParser(ObjectReadContext.empty())
+          .asInstanceOf[JsonParser with ByteBufferFeeder]
+        try {
+          bs match {
+            case bs: ByteString.ByteStrings =>
+              bs.asByteBuffers.foreach(parser.feedInput)
+            case bytes =>
+              parser.feedInput(bytes.asByteBuffer)
+          }
+          objectMapper.readValue[A](parser)
+        } finally
+          parser.close()
+      }(ec)
+    }
+
+  /**
+    * HTTP entity => `Source[A, _]`
+    *
+    * @tparam A
+    *   type to decode
+    * @return
+    *   unmarshaller for `Source[A, _]`
+    */
+  implicit def sourceUnmarshaller[A: JavaTypeable](implicit
+      support: JsonEntityStreamingSupport = EntityStreamingSupport.json()
+  ): FromEntityUnmarshaller[SourceOf[A]] =
+    Unmarshaller
+      .withMaterializer[HttpEntity, SourceOf[A]] { implicit ec => implicit mat => entity =>
+        def asyncParse(bs: ByteString) =
+          Unmarshal(bs).to[A]
+
+        def ordered =
+          Flow[ByteString].mapAsync(support.parallelism)(asyncParse)
+
+        def unordered =
+          Flow[ByteString].mapAsyncUnordered(support.parallelism)(asyncParse)
+
+        Future.successful {
+          entity.dataBytes
+            .via(support.framingDecoder)
+            .via(if (support.unordered) unordered else ordered)
+        }
+      }
+      .forContentTypes(unmarshallerContentTypes: _*)
+
+  /**
+    * `SourceOf[A]` => HTTP entity
+    *
+    * @tparam A
+    *   type to encode
+    * @return
+    *   marshaller for any `SourceOf[A]` value
+    */
+  implicit def sourceMarshaller[A](implicit
+      objectMapper: ObjectMapper = defaultObjectMapper,
+      support: JsonEntityStreamingSupport = EntityStreamingSupport.json()
+  ): ToEntityMarshaller[SourceOf[A]] =
+    jsonSourceStringMarshaller.compose(jsonSource[A])
+}
diff --git a/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/ExampleApp.scala b/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/ExampleApp.scala
new file mode 100644
index 0000000..835a445
--- /dev/null
+++ b/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/ExampleApp.scala
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2015 Heiko Seeberger
+ *
+ * Licensed 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 com.github.pjfanning.pekkohttpjackson3
+
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.scaladsl.Http
+import org.apache.pekko.http.scaladsl.model.HttpRequest
+import org.apache.pekko.http.scaladsl.server.Directives
+import org.apache.pekko.http.scaladsl.unmarshalling.Unmarshal
+import org.apache.pekko.stream.scaladsl.Source
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import scala.io.StdIn
+
+object ExampleApp {
+
+  final case class Foo(bar: String)
+
+  def main(args: Array[String]): Unit = {
+    implicit val system: ActorSystem = ActorSystem()
+
+    // provide an implicit ObjectMapper if you want serialization/deserialization to use it
+    // instead of a default ObjectMapper configured only with DefaultScalaModule provided
+    // by JacksonSupport
+    //
+    // for example:
+    //
+    // implicit val objectMapper = new ObjectMapper()
+    //   .registerModule(DefaultScalaModule)
+    //   .registerModule(new GuavaModule())
+
+    Http().newServerAt("127.0.0.1", 8000).bindFlow(route)
+
+    StdIn.readLine("Hit ENTER to exit")
+    Await.ready(system.terminate(), Duration.Inf)
+  }
+
+  def route(implicit sys: ActorSystem) = {
+    import Directives._
+    import JacksonSupport._
+
+    pathSingleSlash {
+      post {
+
+        entity(as[Foo]) { foo =>
+          complete {
+            foo
+          }
+        }
+      }
+    } ~ pathPrefix("stream") {
+      post {
+        entity(as[SourceOf[Foo]]) { (fooSource: SourceOf[Foo]) =>
+          complete(fooSource.throttle(1, 2.seconds))
+        }
+      } ~ get {
+        pathEndOrSingleSlash {
+          complete(
+            Source(0 to 5)
+              .throttle(1, 1.seconds)
+              .map(i => Foo(s"bar-$i"))
+          )
+        } ~ pathPrefix("remote") {
+          onSuccess(Http().singleRequest(HttpRequest(uri = "http://localhost:8000/stream"))) {
+            response => complete(Unmarshal(response).to[SourceOf[Foo]])
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupportSpec.scala b/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupportSpec.scala
new file mode 100644
index 0000000..114689b
--- /dev/null
+++ b/pekko-http-jackson3/src/test/scala/com/github/pjfanning/pekkohttpjackson3/JacksonSupportSpec.scala
@@ -0,0 +1,179 @@
+/*
+ * Copyright 2015 Heiko Seeberger
+ *
+ * Licensed 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 com.github.pjfanning.pekkohttpjackson3
+
+import tools.jackson.core.StreamReadFeature
+import tools.jackson.core.util.JsonRecyclerPools.BoundedPool
+import tools.jackson.databind.json.JsonMapper
+import tools.jackson.module.scala.DefaultScalaModule
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.scaladsl.marshalling.Marshal
+import org.apache.pekko.http.scaladsl.model._
+import org.apache.pekko.http.scaladsl.model.ContentTypes.{ `application/json`, `text/plain(UTF-8)` }
+import org.apache.pekko.http.scaladsl.unmarshalling.{ Unmarshal, Unmarshaller }
+import org.apache.pekko.http.scaladsl.unmarshalling.Unmarshaller.UnsupportedContentTypeException
+import org.apache.pekko.stream.scaladsl.{ Sink, Source }
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AsyncWordSpec
+
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+
+object JacksonSupportSpec {
+
+  final case class Foo(bar: String) {
+    require(bar startsWith "bar", "bar must start with 'bar'!")
+  }
+}
+
+final class JacksonSupportSpec extends AsyncWordSpec with Matchers with BeforeAndAfterAll {
+  import JacksonSupport._
+  import JacksonSupportSpec._
+
+  private implicit val system: ActorSystem = ActorSystem()
+
+  "JacksonSupport" should {
+    "should enable marshalling and unmarshalling of case classes" in {
+      val foo = Foo("bar")
+      Marshal(foo)
+        .to[RequestEntity]
+        .flatMap(Unmarshal(_).to[Foo])
+        .map(_ shouldBe foo)
+    }
+
+    "enable streamed marshalling and unmarshalling for json arrays" in {
+      val foos = (0 to 100).map(i => Foo(s"bar-$i")).toList
+
+      Marshal(Source(foos))
+        .to[RequestEntity]
+        .flatMap(entity => Unmarshal(entity).to[SourceOf[Foo]])
+        .flatMap(_.runWith(Sink.seq))
+        .map(_ shouldBe foos)
+    }
+
+    "should enable marshalling and unmarshalling of arrays of values" in {
+      val foo = Seq(Foo("bar"))
+      Marshal(foo)
+        .to[RequestEntity]
+        .flatMap(Unmarshal(_).to[Seq[Foo]])
+        .map(_ shouldBe foo)
+    }
+
+    "provide proper error messages for requirement errors" in {
+      val entity = HttpEntity(MediaTypes.`application/json`, """{ "bar": "baz" }""")
+      Unmarshal(entity)
+        .to[Foo]
+        .failed
+        .map(_.getMessage should include("requirement failed: bar must start with 'bar'!"))
+    }
+
+    "fail with NoContentException when unmarshalling empty entities" in {
+      val entity = HttpEntity.empty(`application/json`)
+      Unmarshal(entity)
+        .to[Foo]
+        .failed
+        .map(_ shouldBe Unmarshaller.NoContentException)
+    }
+
+    "fail with UnsupportedContentTypeException when Content-Type is not `application/json`" in {
+      val entity = HttpEntity("""{ "bar": "bar" }""")
+      Unmarshal(entity)
+        .to[Foo]
+        .failed
+        .map(
+          _ shouldBe UnsupportedContentTypeException(Some(`text/plain(UTF-8)`), `application/json`)
+        )
+    }
+
+    "allow unmarshalling with passed in Content-Types" in {
+      val foo = Foo("bar")
+      val `application/json-home` =
+        MediaType.applicationWithFixedCharset("json-home", HttpCharsets.`UTF-8`, "json-home")
+
+      final object CustomJacksonSupport extends JacksonSupport {
+        override def unmarshallerContentTypes = List(`application/json`, `application/json-home`)
+      }
+      import CustomJacksonSupport._
+
+      val entity = HttpEntity(`application/json-home`, """{ "bar": "bar" }""")
+      Unmarshal(entity).to[Foo].map(_ shouldBe foo)
+    }
+
+    "default the stream read constraints" in {
+      val defaultFactory = JacksonSupport.createJsonFactory(JacksonSupport.jacksonConfig)
+      val src            = defaultFactory.streamReadConstraints()
+      src.getMaxNestingDepth shouldEqual 1000
+      src.getMaxNumberLength shouldEqual 1000
+      src.getMaxStringLength shouldEqual 20000000
+      src.getMaxNameLength shouldEqual 50000
+      src.getMaxDocumentLength shouldEqual -1
+    }
+
+    "default the stream write constraints" in {
+      val defaultFactory = JacksonSupport.createJsonFactory(JacksonSupport.jacksonConfig)
+      val swc            = defaultFactory.streamWriteConstraints()
+      swc.getMaxNestingDepth shouldEqual 1000
+    }
+
+    "default the buffer recycler" in {
+      val defaultFactory = JacksonSupport.createJsonFactory(JacksonSupport.jacksonConfig)
+      val pool           = defaultFactory._getRecyclerPool()
+      pool.getClass.getSimpleName shouldEqual "ThreadLocalPool"
+    }
+
+    "support config override for the buffer recycler" in {
+      val testCfg = ConfigFactory
+        .parseString("""buffer-recycler.pool-instance=bounded
+                       |buffer-recycler.bounded-pool-size=1234""".stripMargin)
+        .withFallback(JacksonSupport.jacksonConfig)
+      val factory = JacksonSupport.createJsonFactory(testCfg)
+      val pool    = factory._getRecyclerPool()
+      pool.getClass.getSimpleName shouldEqual "BoundedPool"
+      pool.asInstanceOf[BoundedPool].capacity() shouldEqual 1234
+    }
+
+    "respect pekko-http-json.jackson.read.feature.include-source-in-location" in {
+      val defaultFactory = JacksonSupport.createJsonFactory(JacksonSupport.jacksonConfig)
+      defaultFactory.isEnabled(StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION) shouldBe false
+      val testCfg = ConfigFactory
+        .parseString("read.feature.include-source-in-location=true")
+        .withFallback(JacksonSupport.jacksonConfig)
+      val testFactory = JacksonSupport.createJsonFactory(testCfg)
+      testFactory.isEnabled(StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION) shouldBe true
+      JsonMapper
+        .builder(testFactory)
+        .build()
+        .isEnabled(StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION) shouldBe true
+    }
+
+    "support loading DefaultScalaModule" in {
+      val testCfg = JacksonSupport.jacksonConfig
+      val mapper  = JacksonSupport.createObjectMapper(testCfg)
+      import org.apache.pekko.util.ccompat.JavaConverters._
+      mapper.getRegisteredModules.asScala.map(_.getClass) should contain(
+        classOf[DefaultScalaModule]
+      )
+    }
+  }
+
+  override protected def afterAll() = {
+    Await.ready(system.terminate(), 42.seconds)
+    super.afterAll()
+  }
+}