chore: use Hub

Author: He-Pin

stream/src/main/scala/org/apache/pekko/stream/impl/ActorProcessor.scala

@@ -22,41 +22,7 @@ import pekko.stream.ActorAttributes
 import pekko.stream.impl.ActorSubscriberMessage.{ OnComplete, OnError, OnNext, OnSubscribe }
 import pekko.util.unused
 
-import org.reactivestreams.{ Processor, Subscriber, Subscription }
-
-/**
- * INTERNAL API
- */
-@InternalApi private[pekko] object ActorProcessor {
-
-  def apply[I, O](impl: ActorRef): ActorProcessor[I, O] = {
-    val p = new ActorProcessor[I, O](impl)
-    // Resolve cyclic dependency with actor. This MUST be the first message no matter what.
-    impl ! ExposedPublisher(p.asInstanceOf[ActorPublisher[Any]])
-    p
-  }
-}
-
-/**
- * INTERNAL API
- */
-@InternalApi private[pekko] class ActorProcessor[I, O](impl: ActorRef)
-    extends ActorPublisher[O](impl)
-    with Processor[I, O] {
-  override def onSubscribe(s: Subscription): Unit = {
-    ReactiveStreamsCompliance.requireNonNullSubscription(s)
-    impl ! OnSubscribe(s)
-  }
-  override def onError(t: Throwable): Unit = {
-    ReactiveStreamsCompliance.requireNonNullException(t)
-    impl ! OnError(t)
-  }
-  override def onComplete(): Unit = impl ! OnComplete
-  override def onNext(elem: I): Unit = {
-    ReactiveStreamsCompliance.requireNonNullElement(elem)
-    impl ! OnNext(elem)
-  }
-}
+import org.reactivestreams.{ Subscriber, Subscription }
 
 /**
  * INTERNAL API

stream/src/main/scala/org/apache/pekko/stream/impl/FanoutProcessor.scala

@@ -108,24 +108,6 @@ import org.reactivestreams.Subscriber
     new PublisherSink[In](attr, amendShape(attr))
 }
 
-/**
- * INTERNAL API
- */
-@InternalApi private[pekko] final class FanoutPublisherSink[In](val attributes: Attributes, shape: SinkShape[In])
-    extends SinkModule[In, Publisher[In]](shape) {
-
-  override def create(context: MaterializationContext): (Subscriber[In], Publisher[In]) = {
-    val impl = context.materializer.actorOf(context, FanoutProcessorImpl.props(context.effectiveAttributes))
-    val fanoutProcessor = new ActorProcessor[In, In](impl)
-    // Resolve cyclic dependency with actor. This MUST be the first message no matter what.
-    impl ! ExposedPublisher(fanoutProcessor.asInstanceOf[ActorPublisher[Any]])
-    (fanoutProcessor, fanoutProcessor)
-  }
-
-  override def withAttributes(attr: Attributes): SinkModule[In, Publisher[In]] =
-    new FanoutPublisherSink[In](attr, amendShape(attr))
-}
-
 /**
  * INTERNAL API
  * Attaches a subscriber to this stream.