Creating a network service discovery request

[IvBaranov]

So I have an observable singleton:

/** Provides {@link BonjourEvent} observable. */
  @Provides @Singleton Observable<BonjourEvent> provideBonjourEventStream(Application app) {
    return RxBonjour.startDiscovery(app, "_myservers._tcp");
  }

"myservers" never changes.
In my fragment I have:

@Override public void onStart() {
    super.onStart();
    subscription = bonjourEventStream  
        .subscribeOn(Schedulers.computation()) //
        .observeOn(AndroidSchedulers.mainThread()) //
        .subscribe(bonjourEvent -> {
            ...
         });
}

@Override public void onPause() {
    super.onPause();
    if (subscription != null && !subscription.isUnsubscribed()) subscription.unsubscribe();
  }

So when I switch the screen on an off the subscription unsubscribes and subscribes again. One out of ten times of switching the screen I have:

FATAL EXCEPTION: RxCachedThreadScheduler-15
E/AndroidRuntime: Process: com.symway.assist, PID: 24660
E/AndroidRuntime: java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
E/AndroidRuntime:     at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:60)
E/AndroidRuntime:     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
E/AndroidRuntime:     at java.util.concurrent.FutureTask.run(FutureTask.java:237)
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
E/AndroidRuntime:     at java.lang.Thread.run(Thread.java:841)
E/AndroidRuntime:  Caused by: rx.exceptions.OnErrorNotImplementedException: MulticastLock under-locked RxBonjourDiscovery
E/AndroidRuntime:     at rx.Observable$29.onError(Observable.java:7498)
E/AndroidRuntime:     at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:154)
E/AndroidRuntime:     at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:111)
E/AndroidRuntime:     at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.pollQueue(OperatorObserveOn.java:197)
E/AndroidRuntime:     at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber$2.call(OperatorObserveOn.java:170)
E/AndroidRuntime:     at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
E/AndroidRuntime:     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422) 
E/AndroidRuntime:     at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152) 
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265) 
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112) 
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587) 
E/AndroidRuntime:     at java.lang.Thread.run(Thread.java:841) 
E/AndroidRuntime:  Caused by: java.lang.RuntimeException: MulticastLock under-locked RxBonjourDiscovery
E/AndroidRuntime:     at android.net.wifi.WifiManager$MulticastLock.release(WifiManager.java:2132)
E/AndroidRuntime:     at rxbonjour.internal.SupportBonjourDiscovery$2$1.call(SupportBonjourDiscovery.java:171)
E/AndroidRuntime:     at rxbonjour.internal.SupportBonjourDiscovery$2$1.call(SupportBonjourDiscovery.java:163)
E/AndroidRuntime:     at rx.Observable.unsafeSubscribe(Observable.java:7713)
E/AndroidRuntime:     at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62)
E/AndroidRuntime:     at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) 
E/AndroidRuntime:     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422) 
E/AndroidRuntime:     at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152) 
E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265) 
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112) 
E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587) 
E/AndroidRuntime:     at java.lang.Thread.run(Thread.java:841) 

On SupportBonjourDiscovery.java:171 I see the release, so when I try to subscribe again on screen on it SHOULD go wrong but sometimes it doesn't.

Finally, I have got a question if one just shouldn't unsubscribe ever or one must not have singleton BonjourEvent observable?

[mannodermaus]

I have been able to reproduce this issue. The gist of it boils down to JmDNS.close() being a method that blocks the current thread for quite a while upon cleaning up after a Support-Implementation subscription is being unsubscribed from. Because the Multicast lock previously acquired is only released after this blocking method finishes, this code may execute right before you can acquire a new wake lock.

Releasing the wake lock before the synchronously blocking close() on JmDNS fixes the issue. Also, I have changed the support implementation to only have one multicast lock.

I want to perform some more testing on this and release this in 0.3.1 shortly, if everything is working as intended!

[IvBaranov]

Thank you for explanation.

[mannodermaus]

By the way, I don't see anything wrong with you using and consuming the Observable as a Singleton provided by Dagger, that is totally fine!

[mannodermaus]

Released in 0.3.1