raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

=str IO stream sink/source materialized value must be boxed explicitly

Browse source
This commit is contained in:
Konrad Malawski 2015-04-24 20:22:16 +02:00
parent 6d08cd48c6
commit 60f0bfba5d
5 changed files with 18 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

7
akka-docs-dev/rst/scala/code/docs/stream/io/StreamFileDocSpec.scala
View file

@ -14,6 +14,8 @@ import akka.stream.testkit.Utils._
import akka.stream.testkit._ import akka.stream.testkit._
import akka.util.ByteString import akka.util.ByteString
import scala.concurrent.Future
class StreamFileDocSpec extends AkkaSpec(UnboundedMailboxConfig) { class StreamFileDocSpec extends AkkaSpec(UnboundedMailboxConfig) {
implicit val ec = system.dispatcher implicit val ec = system.dispatcher
@ -45,8 +47,9 @@ class StreamFileDocSpec extends AkkaSpec(UnboundedMailboxConfig) {
//#file-source //#file-source
SynchronousFileSource(file) val foreach: Future[Long] = SynchronousFileSource(file)
.runForeach((chunk: ByteString) ⇒ handle(chunk)) .to(Sink.ignore)
.run()
//#file-source //#file-source
} }

6
akka-stream/src/main/scala/akka/stream/io/InputStreamSource.scala
View file

@ -36,7 +36,7 @@ object InputStreamSource {
* *
* It materializes a [[Future]] containing the number of bytes read from the source file upon completion. * It materializes a [[Future]] containing the number of bytes read from the source file upon completion.
*/ */
def create(createInputStream: Creator[InputStream]): javadsl.Source[ByteString, Future[Long]] = def create(createInputStream: Creator[InputStream]): javadsl.Source[ByteString, Future[java.lang.Long]] =
create(createInputStream, DefaultChunkSize) create(createInputStream, DefaultChunkSize)
/** /**
@ -47,7 +47,7 @@ object InputStreamSource {
* *
* It materializes a [[Future]] containing the number of bytes read from the source file upon completion. * It materializes a [[Future]] containing the number of bytes read from the source file upon completion.
*/ */
def create(createInputStream: Creator[InputStream], chunkSize: Int): javadsl.Source[ByteString, Future[Long]] = def create(createInputStream: Creator[InputStream], chunkSize: Int): javadsl.Source[ByteString, Future[java.lang.Long]] =
apply(() ⇒ createInputStream.create(), chunkSize).asJava apply(() ⇒ createInputStream.create(), chunkSize).asJava.asInstanceOf[javadsl.Source[ByteString, Future[java.lang.Long]]]
} }

4
akka-stream/src/main/scala/akka/stream/io/OutputStreamSink.scala
View file

@ -38,7 +38,7 @@ object OutputStreamSink {
* *
* Materializes a [[Future]] that will be completed with the size of the file (in bytes) at the streams completion. * Materializes a [[Future]] that will be completed with the size of the file (in bytes) at the streams completion.
*/ */
def create(f: Creator[OutputStream]): javadsl.Sink[ByteString, Future[Long]] = def create(f: Creator[OutputStream]): javadsl.Sink[ByteString, Future[java.lang.Long]] =
apply(() ⇒ f.create()).asJava apply(() ⇒ f.create()).asJava.asInstanceOf[javadsl.Sink[ByteString, Future[java.lang.Long]]]
} }

10
akka-stream/src/main/scala/akka/stream/io/SynchronousFileSink.scala
View file

@ -28,7 +28,7 @@ object SynchronousFileSink {
* unless configured otherwise by using [[ActorOperationAttributes]]. * unless configured otherwise by using [[ActorOperationAttributes]].
*/ */
def apply(f: File, append: Boolean = false): Sink[ByteString, Future[Long]] = def apply(f: File, append: Boolean = false): Sink[ByteString, Future[Long]] =
new Sink(new SynchronousFileSink(f, append, DefaultAttributes, Sink.shape("SynchronousFileSink"))) new Sink(new impl.SynchronousFileSink(f, append, DefaultAttributes, Sink.shape("SynchronousFileSink")))
/** /**
* Java API * Java API
@ -41,8 +41,8 @@ object SynchronousFileSink {
* This source is backed by an Actor which will use the dedicated `akka.stream.file-io-dispatcher`, * This source is backed by an Actor which will use the dedicated `akka.stream.file-io-dispatcher`,
* unless configured otherwise by using [[ActorOperationAttributes]]. * unless configured otherwise by using [[ActorOperationAttributes]].
*/ */
def create(f: File): javadsl.Sink[ByteString, Future[Long]] = def create(f: File): javadsl.Sink[ByteString, Future[java.lang.Long]] =
apply(f, append = false).asJava apply(f, append = false).asJava.asInstanceOf[javadsl.Sink[ByteString, Future[java.lang.Long]]]
/** /**
* Java API * Java API
@ -54,7 +54,7 @@ object SynchronousFileSink {
* This source is backed by an Actor which will use the dedicated `akka.stream.file-io-dispatcher`, * This source is backed by an Actor which will use the dedicated `akka.stream.file-io-dispatcher`,
* unless configured otherwise by using [[ActorOperationAttributes]]. * unless configured otherwise by using [[ActorOperationAttributes]].
*/ */
def appendTo(f: File): javadsl.Sink[ByteString, Future[Long]] = def appendTo(f: File): javadsl.Sink[ByteString, Future[java.lang.Long]] =
apply(f, append = true).asJava apply(f, append = true).asInstanceOf[javadsl.Sink[ByteString, Future[java.lang.Long]]]
} }

6
akka-stream/src/main/scala/akka/stream/io/SynchronousFileSource.scala
View file

@ -41,7 +41,7 @@ object SynchronousFileSource {
* *
* It materializes a [[Future]] containing the number of bytes read from the source file upon completion. * It materializes a [[Future]] containing the number of bytes read from the source file upon completion.
*/ */
def create(f: File): javadsl.Source[ByteString, Future[Long]] = def create(f: File): javadsl.Source[ByteString, Future[java.lang.Long]] =
create(f, DefaultChunkSize) create(f, DefaultChunkSize)
/** /**
@ -54,7 +54,7 @@ object SynchronousFileSource {
* *
* It materializes a [[Future]] containing the number of bytes read from the source file upon completion. * It materializes a [[Future]] containing the number of bytes read from the source file upon completion.
*/ */
def create(f: File, chunkSize: Int): javadsl.Source[ByteString, Future[Long]] = def create(f: File, chunkSize: Int): javadsl.Source[ByteString, Future[java.lang.Long]] =
apply(f, chunkSize).asJava apply(f, chunkSize).asJava.asInstanceOf[javadsl.Source[ByteString, Future[java.lang.Long]]]
} }