StacClient supports pagination and has fs2.Streams in the interface

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ## [Unreleased]
 ### Added
 - Add Scala 2.13 cross compilation [#310](https://github.com/azavea/stac4s/pull/310)
+- STAC Client pagination support [#327](https://github.com/azavea/stac4s/pull/327)
 
 ### Changed
 - Make StacClient type alias a trait [#325](https://github.com/azavea/stac4s/pull/325)

build.sbt

@@ -1,5 +1,4 @@
 import xerial.sbt.Sonatype._
-import Dependencies._
 
 lazy val commonSettings = Seq(
   // We are overriding the default behavior of sbt-git which, by default, only
@@ -104,22 +103,20 @@ lazy val credentialSettings = Seq(
 
 val jvmGeometryDependencies = Def.setting {
   Seq(
-    "org.locationtech.jts" % "jts-core" % Versions.Jts,
-    geotrellis("vector").value
+    "org.locationtech.jts"         % "jts-core"          % Versions.Jts,
+    "org.locationtech.geotrellis" %% "geotrellis-vector" % Versions.GeoTrellis.value
   )
 }
 
 val coreDependenciesJVM = Def.setting {
-  Seq(
-    "org.threeten" % "threeten-extra" % Versions.ThreeTenExtra
-  ) ++ jvmGeometryDependencies.value
+  Seq("org.threeten" % "threeten-extra" % Versions.ThreeTenExtra) ++ jvmGeometryDependencies.value
 }
 
 val testingDependenciesJVM = Def.setting {
   Seq(
-    geotrellis("vector").value,
-    "org.locationtech.jts" % "jts-core"       % Versions.Jts,
-    "org.threeten"         % "threeten-extra" % Versions.ThreeTenExtra
+    "org.locationtech.geotrellis" %% "geotrellis-vector" % Versions.GeoTrellis.value,
+    "org.locationtech.jts"         % "jts-core"          % Versions.Jts,
+    "org.threeten"                 % "threeten-extra"    % Versions.ThreeTenExtra
   )
 }
 
@@ -131,7 +128,7 @@ val testRunnerDependenciesJVM = Seq(
 
 lazy val root = project
   .in(file("."))
-  .settings(moduleName := "root")
+  .settings(name := "stac4s")
   .settings(commonSettings)
   .settings(publishSettings)
   .settings(noPublishSettings)
@@ -182,11 +179,7 @@ lazy val testing = crossProject(JSPlatform, JVMPlatform)
     )
   )
   .jvmSettings(libraryDependencies ++= testingDependenciesJVM.value)
-  .jsSettings(
-    libraryDependencies ++= Seq(
-      "io.github.cquiroz" %%% "scala-java-time" % "2.2.2" % Test
-    )
-  )
+  .jsSettings(libraryDependencies += "io.github.cquiroz" %%% "scala-java-time" % "2.2.2" % Test)
 
 lazy val testingJVM = testing.jvm
 lazy val testingJS  = testing.js
@@ -203,11 +196,7 @@ lazy val coreTest = crossProject(JSPlatform, JVMPlatform)
       "org.scalatestplus" %%% "scalacheck-1-14" % Versions.ScalatestPlusScalacheck % Test
     )
   )
-  .jsSettings(
-    libraryDependencies ++= Seq(
-      "io.github.cquiroz" %%% "scala-java-time" % "2.2.2" % Test
-    )
-  )
+  .jsSettings(libraryDependencies += "io.github.cquiroz" %%% "scala-java-time" % "2.2.2" % Test)
 
 lazy val coreTestJVM = coreTest.jvm
 lazy val coreTestJS  = coreTest.js
@@ -232,6 +221,7 @@ lazy val client = crossProject(JSPlatform, JVMPlatform)
       "com.softwaremill.sttp.client3" %%% "json-common"   % Versions.Sttp,
       "com.softwaremill.sttp.model"   %%% "core"          % Versions.SttpModel,
       "com.softwaremill.sttp.shared"  %%% "core"          % Versions.SttpShared,
+      "co.fs2"                        %%% "fs2-core"      % Versions.Fs2,
       "org.scalatest"                 %%% "scalatest"     % Versions.Scalatest % Test
     )
   )

modules/client/js/src/main/scala/com/azavea/stac4s/api/client/SttpStacClient.scala

@@ -1,11 +1,11 @@
 package com.azavea.stac4s.api.client
 
-import cats.MonadError
+import cats.MonadThrow
 import sttp.client3.SttpBackend
 import sttp.model.Uri
 
 object SttpStacClient {
 
-  def apply[F[_]: MonadError[*[_], Throwable]](client: SttpBackend[F, Any], baseUri: Uri): SttpStacClient[F] =
+  def apply[F[_]: MonadThrow](client: SttpBackend[F, Any], baseUri: Uri): SttpStacClient[F] =
     SttpStacClientF[F, SearchFilters](client, baseUri)
 }

modules/client/jvm/src/main/scala/com/azavea/stac4s/api/client/SttpStacClient.scala

@@ -1,11 +1,11 @@
 package com.azavea.stac4s.api.client
 
-import cats.MonadError
+import cats.MonadThrow
 import sttp.client3.SttpBackend
 import sttp.model.Uri
 
 object SttpStacClient {
 
-  def apply[F[_]: MonadError[*[_], Throwable]](client: SttpBackend[F, Any], baseUri: Uri): SttpStacClient[F] =
+  def apply[F[_]: MonadThrow](client: SttpBackend[F, Any], baseUri: Uri): SttpStacClient[F] =
     SttpStacClientF[F, SearchFilters](client, baseUri)
 }

modules/client/shared/src/main/scala/com/azavea/stac4s/api/client/StacClientF.scala

@@ -3,13 +3,14 @@ package com.azavea.stac4s.api.client
 import com.azavea.stac4s._
 
 import eu.timepit.refined.types.string.NonEmptyString
+import fs2.Stream
 
 trait StacClientF[F[_], S] {
-  def search: F[List[StacItem]]
-  def search(filter: S): F[List[StacItem]]
-  def collections: F[List[StacCollection]]
+  def search: Stream[F, StacItem]
+  def search(filter: S): Stream[F, StacItem]
+  def collections: Stream[F, StacCollection]
   def collection(collectionId: NonEmptyString): F[StacCollection]
-  def items(collectionId: NonEmptyString): F[List[StacItem]]
+  def items(collectionId: NonEmptyString): Stream[F, StacItem]
   def item(collectionId: NonEmptyString, itemId: NonEmptyString): F[StacItem]
   def itemCreate(collectionId: NonEmptyString, item: StacItem): F[StacItem]
   def collectionCreate(collection: StacCollection): F[StacCollection]

modules/client/shared/src/main/scala/com/azavea/stac4s/api/client/SttpStacClientF.scala

@@ -1,42 +1,70 @@
 package com.azavea.stac4s.api.client
 
-import com.azavea.stac4s.{StacCollection, StacItem}
+import com.azavea.stac4s.{StacCollection, StacItem, StacLink, StacLinkType}
 
-import cats.MonadError
+import cats.MonadThrow
+import cats.syntax.apply._
+import cats.syntax.either._
 import cats.syntax.flatMap._
-import cats.syntax.functor._
 import cats.syntax.option._
 import eu.timepit.refined.types.string.NonEmptyString
+import fs2.Stream
+import io.circe
 import io.circe.syntax._
 import io.circe.{Encoder, Json}
 import sttp.client3.circe.asJson
-import sttp.client3.{SttpBackend, basicRequest}
+import sttp.client3.{ResponseException, SttpBackend, UriContext, basicRequest}
 import sttp.model.Uri
 
-case class SttpStacClientF[F[_]: MonadError[*[_], Throwable], S: Encoder](
+case class SttpStacClientF[F[_]: MonadThrow, S: Encoder](
     client: SttpBackend[F, Any],
     baseUri: Uri
 ) extends StacClientF[F, S] {
-  def search: F[List[StacItem]] = search(None)
 
-  def search(filter: S): F[List[StacItem]] = search(filter.asJson.some)
+  /** Get the next page [[Uri]] from the received [[Json]] body. */
+  private def getNextLink(body: Either[ResponseException[String, circe.Error], Json]): F[Option[Uri]] =
+    body
+      .flatMap {
+        _.hcursor
+          .downField("links")
+          .as[Option[List[StacLink]]]
+          .map(_.flatMap(_.collectFirst { case l if l.rel == StacLinkType.Next => uri"${l.href}" }))
+      }
+      .liftTo[F]
 
-  private def search(filter: Option[Json]): F[List[StacItem]] =
-    client
-      .send {
-        filter
-          .fold(basicRequest)(f => basicRequest.body(f.asJson.noSpaces))
-          .post(baseUri.withPath("search"))
-          .response(asJson[Json])
+  def search: Stream[F, StacItem] = search(None)
+
+  def search(filter: S): Stream[F, StacItem] = search(filter.asJson.some)
+
+  private def search(filter: Option[Json]): Stream[F, StacItem] =
+    Stream
+      .unfoldLoopEval(baseUri.withPath("search")) { link =>
+        client
+          .send(filter.fold(basicRequest)(f => basicRequest.body(f.asJson.noSpaces)).post(link).response(asJson[Json]))
+          .flatMap { response =>
+            val body     = response.body
+            val items    = body.flatMap(_.hcursor.downField("features").as[List[StacItem]]).liftTo[F]
+            val nextLink = getNextLink(body)
+
+            (items, nextLink).tupled
+          }
       }
-      .map(_.body.flatMap(_.hcursor.downField("features").as[List[StacItem]]))
-      .flatMap(MonadError[F, Throwable].fromEither)
+      .flatMap(Stream.emits)
 
-  def collections: F[List[StacCollection]] =
-    client
-      .send(basicRequest.get(baseUri.withPath("collections")).response(asJson[Json]))
-      .map(_.body.flatMap(_.hcursor.downField("collections").as[List[StacCollection]]))
-      .flatMap(MonadError[F, Throwable].fromEither)
+  def collections: Stream[F, StacCollection] =
+    Stream
+      .unfoldLoopEval(baseUri.withPath("collections")) { link =>
+        client
+          .send(basicRequest.get(link).response(asJson[Json]))
+          .flatMap { response =>
+            val body     = response.body
+            val items    = body.flatMap(_.hcursor.downField("collections").as[List[StacCollection]]).liftTo[F]
+            val nextLink = getNextLink(body)
+
+            (items, nextLink).tupled
+          }
+      }
+      .flatMap(Stream.emits)
 
   def collection(collectionId: NonEmptyString): F[StacCollection] =
     client
@@ -45,14 +73,23 @@ case class SttpStacClientF[F[_]: MonadError[*[_], Throwable], S: Encoder](
           .get(baseUri.withPath("collections", collectionId.value))
           .response(asJson[StacCollection])
       )
-      .map(_.body)
-      .flatMap(MonadError[F, Throwable].fromEither)
+      .flatMap(_.body.liftTo[F])
 
-  def items(collectionId: NonEmptyString): F[List[StacItem]] =
-    client
-      .send(basicRequest.get(baseUri.withPath("collections", collectionId.value, "items")).response(asJson[Json]))
-      .map(_.body.flatMap(_.hcursor.downField("features").as[List[StacItem]]))
-      .flatMap(MonadError[F, Throwable].fromEither)
+  def items(collectionId: NonEmptyString): Stream[F, StacItem] = {
+    Stream
+      .unfoldLoopEval(baseUri.withPath("collections", collectionId.value, "items")) { link =>
+        client
+          .send(basicRequest.get(link).response(asJson[Json]))
+          .flatMap { response =>
+            val body     = response.body
+            val items    = body.flatMap(_.hcursor.downField("features").as[List[StacItem]]).liftTo[F]
+            val nextLink = getNextLink(body)
+
+            (items, nextLink).tupled
+          }
+      }
+      .flatMap(Stream.emits)
+  }
 
   def item(collectionId: NonEmptyString, itemId: NonEmptyString): F[StacItem] =
     client
@@ -61,8 +98,7 @@ case class SttpStacClientF[F[_]: MonadError[*[_], Throwable], S: Encoder](
           .get(baseUri.withPath("collections", collectionId.value, "items", itemId.value))
           .response(asJson[StacItem])
       )
-      .map(_.body)
-      .flatMap(MonadError[F, Throwable].fromEither)
+      .flatMap(_.body.liftTo[F])
 
   def itemCreate(collectionId: NonEmptyString, item: StacItem): F[StacItem] =
     client
@@ -72,8 +108,7 @@ case class SttpStacClientF[F[_]: MonadError[*[_], Throwable], S: Encoder](
           .body(item.asJson.noSpaces)
           .response(asJson[StacItem])
       )
-      .map(_.body)
-      .flatMap(MonadError[F, Throwable].fromEither)
+      .flatMap(_.body.liftTo[F])
 
   def collectionCreate(collection: StacCollection): F[StacCollection] =
     client
@@ -83,6 +118,5 @@ case class SttpStacClientF[F[_]: MonadError[*[_], Throwable], S: Encoder](
           .body(collection.asJson.noSpaces)
           .response(asJson[StacCollection])
       )
-      .map(_.body)
-      .flatMap(MonadError[F, Throwable].fromEither)
+      .flatMap(_.body.liftTo[F])
 }

modules/client/shared/src/test/scala/com/azavea/stac4s/api/client/SttpEitherInstances.scala

@@ -0,0 +1,33 @@
+package com.azavea.stac4s.api.client
+
+import cats.effect.{ExitCase, Sync}
+
+trait SttpEitherInstances {
+
+  /** [[Sync]] instance defined for Either[Throwable, *].
+    * It is required (sadly) to derive [[fs2.Stream.Compiler]] which is necessary for the [[fs2.Stream.compile]] function.
+    */
+  implicit val eitherSync: Sync[Either[Throwable, *]] = new Sync[Either[Throwable, *]] {
+    lazy val me = cats.instances.either.catsStdInstancesForEither[Throwable]
+
+    def suspend[A](thunk: => Either[Throwable, A]): Either[Throwable, A] = thunk
+
+    def bracketCase[A, B](acquire: Either[Throwable, A])(use: A => Either[Throwable, B])(
+        release: (A, ExitCase[Throwable]) => Either[Throwable, Unit]
+    ): Either[Throwable, B] =
+      flatMap(acquire)(use)
+
+    def flatMap[A, B](fa: Either[Throwable, A])(f: A => Either[Throwable, B]): Either[Throwable, B] =
+      fa.flatMap(f)
+
+    def tailRecM[A, B](a: A)(f: A => Either[Throwable, Either[A, B]]): Either[Throwable, B] =
+      me.tailRecM(a)(f)
+
+    def raiseError[A](e: Throwable): Either[Throwable, A] = me.raiseError(e)
+
+    def handleErrorWith[A](fa: Either[Throwable, A])(f: Throwable => Either[Throwable, A]): Either[Throwable, A] =
+      me.handleErrorWith(fa)(f)
+
+    def pure[A](x: A): Either[Throwable, A] = me.pure(x)
+  }
+}

modules/client/shared/src/test/scala/com/azavea/stac4s/api/client/SttpStacClientFSpec.scala

@@ -16,14 +16,15 @@ import sttp.client3.testing.SttpBackendStub
 import sttp.model.Method
 import sttp.monad.EitherMonad
 
-trait SttpStacClientFSpec[S] extends AnyFunSpec with Matchers with BeforeAndAfterAll {
+trait SttpStacClientFSpec[S] extends AnyFunSpec with Matchers with BeforeAndAfterAll with SttpEitherInstances {
 
   def arbCollectionShort: Arbitrary[StacCollection]
   def arbItemCollectionShort: Arbitrary[ItemCollection]
   def arbItemShort: Arbitrary[StacItem]
 
   def client: SttpStacClientF[Either[Throwable, *], S]
 
+  /** We use the default synchronous Either backend to use the same tests set for the Scala JS backend. */
   lazy val backend =
     SttpBackendStub(EitherMonad)
       .whenRequestMatches(_.uri.path == Seq("search"))
@@ -86,31 +87,37 @@ trait SttpStacClientFSpec[S] extends AnyFunSpec with Matchers with BeforeAndAfte
 
   describe("SttpStacClientSpec") {
     it("search") {
-      client.search
+      client.search.compile.toList
         .valueOr(throw _)
     }
 
     it("collections") {
-      client.collections
+      client.collections.compile.toList
         .valueOr(throw _)
+        .map(_.id should not be empty)
     }
 
     it("collection") {
       client
         .collection(NonEmptyString.unsafeFrom("collection_id"))
         .valueOr(throw _)
+        .id should not be empty
     }
 
     it("items") {
       client
         .items(NonEmptyString.unsafeFrom("collection_id"))
+        .compile
+        .toList
         .valueOr(throw _)
+        .map(_.id should not be empty)
     }
 
     it("item") {
       client
         .item(NonEmptyString.unsafeFrom("collection_id"), NonEmptyString.unsafeFrom("item_id"))
         .valueOr(throw _)
+        .id should not be empty
     }
 
     it("itemCreate") {