Hi Jason, On Mon, Jun 1, 2015 at 8:24 PM, Jason Martens <[email protected]> wrote:
> Hello All, > I deal with a lot of large files on Amazon S3 and disk, which have been > indexed into chunks that can be used individually. What I would like to do > is have a Source of offsets and sizes which can be filtered to then pass > into another stream element which could read that particular file from S3 > or from a local disk. However, my attempts thus far to make this work have > failed, since the stream element which does the reading seems to block > after the first read. > What do you mean it is blocked? What is the code that fails? Can you give a small example? > In effect, I guess this is equivalent to having an upstream and downstream > source, which is perhaps not supported. > Umm, what does a downstream Source mean? Sources have exactly one output port, so they cannot be downstream. If you want to have a stream processing element that takes input, then it must be a Sink, Flow, BidiFlow (or a more complex graph) that has at least one input port. Alternatively, you can take the stream of indices and offsets, map that stream to a Sources (so now you have a stream of Sources) then use flatten to create one contiguous stream from them. > Does anyone have any suggestions for how I could accomplish this? The key > feature is the ability to select a subset of the index records so that the > entire file does not have to be read from S3. In my imagination it would > look something like this: > > val indexSource[(Long, Long), Unit] = Source(() => index.iterator) > def readFileRange(offset: Long, size: Long): Array[Byte] > > indexSource.skip(10).map(readFileRange).map(println).runWith(Sink.ignore) > // Prints out a byte array for every 10th chunk in the file > indexSource.skip(10).map(fileRangeSource(_)).flatten(FlattenStrategy.concat).runForeach(println) You will need to implement fileRangeSource() to return a Source of bytestrings given the offsets and lengths. -Endre > > Thanks, > Jason > > -- > >>>>>>>>>> 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 http://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Typesafe - Reactive apps on the JVM Blog: letitcrash.com Twitter: @akkateam -- >>>>>>>>>> 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 http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
