fix: not found error in stream queries (#27)

Author: carlos-verdes

.gitignore

@@ -90,3 +90,4 @@ project/metals.sbt
 coursier
 
 *.semanticdb
+nohup.*

arangodb-http/src/it/scala/io/funkode/arangodb/http/ArangoJsonIT.scala

@@ -73,14 +73,15 @@ trait ArangoExamples:
   val geographyCollection2 = CollectionName("geography2")
   val countries = CollectionName("countries")
   val continents = CollectionName("continents")
+  val continents2 = CollectionName("continents2")
   val oceans = CollectionName("oceans")
   val oceans2 = CollectionName("oceans2")
   val graphEdgeDefinitions = List(GraphEdgeDefinition(allies, List(countries), List(countries)))
   val graphEdgeDefinitionsGeography = List(
     GraphEdgeDefinition(geographyCollection, List(continents), List(continents))
   )
   val graphEdgeDefinitionsGeography2 = List(
-    GraphEdgeDefinition(geographyCollection2, List(continents), List(continents))
+    GraphEdgeDefinition(geographyCollection2, List(continents2), List(continents2))
   )
   val es = DocumentHandle(countries, DocumentKey("ES"))
   val fr = DocumentHandle(countries, DocumentKey("FR"))
@@ -205,17 +206,30 @@ object ArangoJsonIT extends ZIOSpecDefault with ArangoExamples:
             .read[Pet]()
             .flip
             .debug("test collection doesn't exist")
-        yield assertTrue(error.code == 404L)
+        yield assertTrue(
+          error == ArangoError(404L, true, "collection or view not found: collectionDoesntExist", 1203L)
+        )
       },
       test("Manage document doesn't exist when fetching document") {
         for
-          invalidCollection <- ArangoClientJson.collection(CollectionName("countries"))
-          error <- invalidCollection
+          countriesCollection <- ArangoClientJson.collection(CollectionName("countries"))
+          error <- countriesCollection
             .document(DocumentKey("turtle"))
             .read[Pet]()
             .flip
             .debug("test document doesn't exist")
-        yield assertTrue(error.code == 404L)
+        yield assertTrue(error == ArangoError(404L, true, "document not found", 1202L))
+      },
+      test("Manage document doesn't exist when fetching raw documents") {
+        for
+          countriesCollection <- ArangoClientJson.collection(CollectionName("countries"))
+          error <- countriesCollection
+            .document(DocumentKey("turtle"))
+            .readRaw()
+            .flatMap(_.via(ZPipeline.utf8Decode).run(ZSink.collectAll))
+            .flip
+            .debug("test document raw doesn't exist")
+        yield assertTrue(error == ArangoError(404L, true, "document not found", 1202L))
       },
       test("Query documents with cursor") {
         for
@@ -280,7 +294,7 @@ object ArangoJsonIT extends ZIOSpecDefault with ArangoExamples:
           _ <- graph.removeVertexCollection(oceans2)
           collectionsAfter <- graph.vertexCollections
         yield assertTrue(collectionsBefore != collectionsAfter) && assertTrue(
-          collectionsAfter == List(continents)
+          collectionsAfter == List(continents2)
         )
       },
       test("Save and retrieve document from byte array stream") {

arangodb-http/src/main/scala/io/funkode/arangodb/http/ArangoClientHttp.scala

@@ -81,8 +81,10 @@ class ArangoClientHttp[Encoder[_], Decoder[_]](
     yield response
 
   def getRaw(header: ArangoMessage.Header): AIO[ArangoStreamRaw] =
-    for response <- httpClient.request(header.emptyRequest(BaseUrl, headers)).handleErrors
-    yield response.body.asStream.handleStreamErrors
+    for
+      response <- httpClient.request(header.emptyRequest(BaseUrl, headers)).handleErrors
+      parsedResponse <- parseResponseBodyRaw(response)
+    yield parsedResponse
 
   def get[O: Decoder](header: ArangoMessage.Header): AIO[ArangoMessage[O]] =
     for
@@ -95,8 +97,10 @@ class ArangoClientHttp[Encoder[_], Decoder[_]](
   ): AIO[ArangoStreamRaw] =
     val header = message.header.emptyRequest(BaseUrl, headers)
     val request: Request = header.copy(body = Body.fromStream(message.body))
-    for response <- httpClient.request(request).handleErrors
-    yield response.body.asStream.handleStreamErrors
+    for
+      response <- httpClient.request(request).handleErrors
+      parsedResponse <- parseResponseBodyRaw(response)
+    yield parsedResponse
 
   def command[I: Encoder, O: Decoder](message: ArangoMessage[I]): AIO[ArangoMessage[O]] =
     val header = message.header.emptyRequest(BaseUrl, headers)
@@ -124,6 +128,12 @@ class ArangoClientHttp[Encoder[_], Decoder[_]](
     new ArangoClientHttp[Encoder, Decoder](newConfig, httpClient, token)
 
   private def parseResponseBody[O: Decoder](response: Response): AIO[O] =
+    parseResponseBodyInternal(response)(httpDecoder.decode[O])
+
+  private def parseResponseBodyRaw(response: Response): AIO[ArangoStreamRaw] =
+    parseResponseBodyInternal(response)(body => ZIO.succeed(body.asStream.handleStreamErrors))
+
+  private def parseResponseBodyInternal[O](response: Response)(onSucces: Body => AIO[O]): AIO[O] =
     if response.status.isError
     then
       httpDecoder
@@ -149,7 +159,7 @@ class ArangoClientHttp[Encoder[_], Decoder[_]](
           },
           arangoError => ZIO.fail(arangoError)
         )
-    else httpDecoder.decode[O](response.body)
+    else onSucces(response.body)
 
   def currentDatabase: DatabaseName = (config.database)
 
@@ -273,8 +283,8 @@ object extensions:
   extension [A](call: IO[Throwable, A])
     def handleErrors: IO[ArangoError, A] = call.catchAll(e => ZIO.fail(throwableToArangoError(e)))
 
-  extension [A](stream: Stream[Throwable, A])
-    def handleStreamErrors: ArangoStream[A] = stream.catchAll(e => ZStream.fail(throwableToArangoError(e)))
+  extension (stream: Stream[Throwable, Byte])
+    def handleStreamErrors: ArangoStreamRaw = stream.catchAll(e => ZStream.fail(throwableToArangoError(e)))
 
 object ArangoClientSchema: