258         @Override

   258         @Override

   259         public void run() {

   259         public void run() {

   260             try {

   260             try {

   261                 run1();

   261                 run1();

   262             } catch (Throwable t) {

   262             } catch (Throwable t) {

   263                 errorCommon(t);

   265                 errorCommon(t);

   264             }

   266             }

   265         }

   267         }

   266 

   268 

   267         private void run1() {

   269         private void run1() {

   290                 if (debug.on())

   292                 if (debug.on())

   291                     debug.log("DownstreamPusher: forwarding error downstream: " + error);

   293                     debug.log("DownstreamPusher: forwarding error downstream: " + error);

   292                 pushScheduler.stop();

   294                 pushScheduler.stop();

   293                 outputQ.clear();

   295                 outputQ.clear();

   294                 downstreamSubscriber.onError(error);

   296                 downstreamSubscriber.onError(error);

   295                 return;

   298                 return;

   296             }

   299             }

   297 

   300 

   298             // OK - no error, let's proceed

   301             // OK - no error, let's proceed

   299             if (!outputQ.isEmpty()) {

   302             if (!outputQ.isEmpty()) {

   381     protected boolean errorCommon(Throwable throwable) {

   384     protected boolean errorCommon(Throwable throwable) {

   382         assert throwable != null ||

   385         assert throwable != null ||

   383                 (throwable = new AssertionError("null throwable")) != null;

   386                 (throwable = new AssertionError("null throwable")) != null;

   384         if (errorRef.compareAndSet(null, throwable)) {

   387         if (errorRef.compareAndSet(null, throwable)) {

   385             if (debug.on()) debug.log("error", throwable);

   388             if (debug.on()) debug.log("error", throwable);

   386             pushScheduler.runOrSchedule();

   390             pushScheduler.runOrSchedule();

   389             return true;

   391             return true;

   390         }

   392         }

   391         return false;

   393         return false;

   392     }

   394     }

   393 

   395