Why do you want to read the entire document into memory?
On Wed, Jan 27, 2016 at 4:19 AM, Ryan Stradling <[email protected]>
wrote:
> Hi all,
> I have the below code. I have a list of document ids that I want to
> return the documents for. However, I want to stream each document's data
> chunk by chunk. So the general structure would be
>
> Seq[DocId] -> Document(DocId, ByteString) -> sink
>
> So for instance for document id X whose document size is 10 bytes and my
> chunk size is 1 byte I would get
> Document(X, ByteString[Byte 0])
> Document(X, ByteString[Byte 1])
> ...
> Document(X, ByteString[Byte 9])
>
> I totally realize that Application.flow function I have is not correct as
> it will only give me the first chunk size. It is meant as a "here is the
> stateful action I would want to do for my flow."
>
>
>
>
> import java.io.{InputStream}
>
> import akka.actor.ActorSystem
> import akka.stream.ActorMaterializer
> import akka.stream.scaladsl.{Flow, Source}
> import akka.util.ByteString
>
>
> object Database {
>
> case class DocId(id: String)
>
> case class Document(id: DocId, bs: ByteString)
>
> trait Database {
> def getDocumentStream(id: DocId): Option[InputStream]
>
> }
>
> object ResourceDatabase extends Database {
> def getDocumentStream(id: DocId): Option[InputStream] = {
> val rs = getClass.getResourceAsStream("/" + id.id + ".json")
> if (rs != null) {
> Some(rs)
> }
> else {
> None
> }
> }
> }
> }
>
> object Application extends App {
>
> import Database._
>
> val source = Source(Vector(DocId("1"), DocId("2")))
>
> def flow(): Flow[DocId, Option[Document], Unit] = {
> val emptyByteString = ByteString("")
> Flow[DocId].map { x =>
> val ds = ResourceDatabase.getDocumentStream(x)
> ds map { y =>
> /* Know this is not correct but would like to know how to do this
> properly */
> val b = new Array[Byte](8192)
> val bytesRead = y.read(b, 0, 8192)
> if (bytesRead >= 0) {
> Document(x, ByteString(b.take(bytesRead)))
> }
> else {
> y.close
> Document(x, emptyByteString)
> }
> }
> }
> }
>
> import scala.concurrent.ExecutionContext.Implicits.global
>
> implicit val system = ActorSystem("application-streams")
> implicit val materializer = ActorMaterializer()
> val f = source.via(flow).runForeach(x => x map {
> println _
> })
>
> f onFailure {
> case t => println("Failed with " + t.getMessage)
> }
> f onSuccess {
> case t => println("Success")
> }
> system.awaitTermination()
> }
>
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Cheers,
√
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
