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

!str #16521 Add ActorRefSink

Browse source 
* also rename the factory for ActorSubscriber props Sink,
  from apply to actorSubscriber
This commit is contained in:
Patrik Nordwall 2015-03-30 14:42:30 +02:00
parent 8f47b6dfcc
commit 946faedd95
11 changed files with 247 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

29
akka-stream/src/main/scala/akka/stream/impl/Sinks.scala
View file

@ -190,13 +190,34 @@ private[akka] final class CancelSink(val attributes: OperationAttributes, shape:
* Creates and wraps an actor into [[org.reactivestreams.Subscriber]] from the given `props`,
* which should be [[akka.actor.Props]] for an [[akka.stream.actor.ActorSubscriber]].
*/
private[akka] final class PropsSink[In](props: Props, val attributes: OperationAttributes, shape: SinkShape[In]) extends SinkModule[In, ActorRef](shape) {
private[akka] final class ActorSubscriberSink[In](props: Props, val attributes: OperationAttributes, shape: SinkShape[In]) extends SinkModule[In, ActorRef](shape) {
override def create(materializer: ActorFlowMaterializerImpl, flowName: String) = {
val subscriberRef = materializer.actorOf(props, name = s"$flowName-props")
val subscriberRef = materializer.actorOf(props, name = s"$flowName-actorSubscriber")
(akka.stream.actor.ActorSubscriber[In](subscriberRef), subscriberRef)
}
override protected def newInstance(shape: SinkShape[In]): SinkModule[In, ActorRef] = new PropsSink[In](props, attributes, shape)
override def withAttributes(attr: OperationAttributes): Module = new PropsSink[In](props, attr, amendShape(attr))
override protected def newInstance(shape: SinkShape[In]): SinkModule[In, ActorRef] = new ActorSubscriberSink[In](props, attributes, shape)
override def withAttributes(attr: OperationAttributes): Module = new ActorSubscriberSink[In](props, attr, amendShape(attr))
}
/**
* INTERNAL API
*/
private[akka] final class ActorRefSink[In](ref: ActorRef, onCompleteMessage: Any,
val attributes: OperationAttributes,
shape: SinkShape[In]) extends SinkModule[In, Unit](shape) {
override def create(materializer: ActorFlowMaterializerImpl, flowName: String) = {
val subscriberRef = materializer.actorOf(
ActorRefSinkActor.props(ref, materializer.settings.maxInputBufferSize, onCompleteMessage),
name = s"$flowName-actorRef")
(akka.stream.actor.ActorSubscriber[In](subscriberRef), ())
}
override protected def newInstance(shape: SinkShape[In]): SinkModule[In, Unit] =
new ActorRefSink[In](ref, onCompleteMessage, attributes, shape)
override def withAttributes(attr: OperationAttributes): Module =
new ActorRefSink[In](ref, onCompleteMessage, attr, amendShape(attr))
}