Rx-java serialization

Question

I am trying to serialize a subscription to send over a network. I am using Scala and doing something like this:

observable.materialize.subscribe{ n : Notification => sendToNetwork(n)}

However, I get errors:

java.io.NotSerializableException: rx.lang.scala.Notification$OnNext

(To be precise, I use Akka and try to send notifications to the remote player. But I think this problem is more general than this).

It seems like he refuses to serialize the class OnNext, which is actually a subclass Notification, which is the inner class of the companion object rx.lang.scala.Notification:

http://rxscala.imtqy.com/scaladoc/#rx.lang.scala.Notification $$ OnNext

... and I think I saw somewhere in the Java documentation that it is not possible to serialize internal non-static classes.

How do I get it right? If so, is this a limitation of the rx-java class hierarchy? Or is there some way around this and serialize Notifications?

Answer 1

How do I get it right?

You can serialize non-static inner classes if the outer class is serializable. But in both Java and Scala, you need to explicitly tell the compiler that the class is serializable (by extension Serializable), and neither rx-java Notificationor rx- scala OnNextare serializable.

Or is there some way around this and serialize notifications?

Akka : http://doc.akka.io/docs/akka/snapshot/scala/serialization.html. Java .

Answer 2

Kontraktor-Reactive-Streams , . Kontraktor, ( ).

public static void remotingRxToRx() {
    Observable<Integer> range = Observable.range(0, 50_000_000);
    Publisher<Integer> pub = RxReactiveStreams.toPublisher(range);

    KxReactiveStreams.get().asRxPublisher(pub)
        .serve(new TCPNIOPublisher().port(3456));

    RateMeasure rm = new RateMeasure("events");

    KxPublisher<Integer> remoteStream =
        KxReactiveStreams.get()
            .connect(Integer.class, new TCPConnectable().host("localhost").port(3456));

    RxReactiveStreams.toObservable(remoteStream)
        .forEach( i -> rm.count() );
}

: https://github.com/RuedigerMoeller/kontraktor/tree/trunk/modules/reactive-streams/src/examples/src/rxstreamserver

( )