UnicastProcessor.smali
.class public final Lio/reactivex/processors/UnicastProcessor;
.super Lio/reactivex/processors/FlowableProcessor;
.source "UnicastProcessor.java"
# annotations
.annotation system Ldalvik/annotation/MemberClasses;
value = {
Lio/reactivex/processors/UnicastProcessor$UnicastQueueSubscription;
}
.end annotation
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">",
"Lio/reactivex/processors/FlowableProcessor<",
"TT;>;"
}
.end annotation
# instance fields
.field final actual:Ljava/util/concurrent/atomic/AtomicReference;
.annotation system Ldalvik/annotation/Signature;
value = {
"Ljava/util/concurrent/atomic/AtomicReference<",
"Lorg/reactivestreams/Subscriber<",
"-TT;>;>;"
}
.end annotation
.end field
.field volatile cancelled:Z
.field final delayError:Z
.field volatile done:Z
.field enableOperatorFusion:Z
.field error:Ljava/lang/Throwable;
.field final onTerminate:Ljava/util/concurrent/atomic/AtomicReference;
.annotation system Ldalvik/annotation/Signature;
value = {
"Ljava/util/concurrent/atomic/AtomicReference<",
"Ljava/lang/Runnable;",
">;"
}
.end annotation
.end field
.field final once:Ljava/util/concurrent/atomic/AtomicBoolean;
.field final queue:Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
.annotation system Ldalvik/annotation/Signature;
value = {
"Lio/reactivex/internal/queue/SpscLinkedArrayQueue<",
"TT;>;"
}
.end annotation
.end field
.field final requested:Ljava/util/concurrent/atomic/AtomicLong;
.field final wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
.annotation system Ldalvik/annotation/Signature;
value = {
"Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription<",
"TT;>;"
}
.end annotation
.end field
# direct methods
.method static constructor <clinit>()V
.registers 1
return-void
.end method
.method constructor <init>(I)V
.registers 4
.param p1, "capacityHint" # I
.line 150
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
const/4 v0, 0x0
const/4 v1, 0x1
invoke-direct {p0, p1, v0, v1}, Lio/reactivex/processors/UnicastProcessor;-><init>(ILjava/lang/Runnable;Z)V
.line 151
return-void
.end method
.method constructor <init>(ILjava/lang/Runnable;)V
.registers 4
.param p1, "capacityHint" # I
.param p2, "onTerminate" # Ljava/lang/Runnable;
.line 161
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
const/4 v0, 0x1
invoke-direct {p0, p1, p2, v0}, Lio/reactivex/processors/UnicastProcessor;-><init>(ILjava/lang/Runnable;Z)V
.line 162
return-void
.end method
.method constructor <init>(ILjava/lang/Runnable;Z)V
.registers 6
.param p1, "capacityHint" # I
.param p2, "onTerminate" # Ljava/lang/Runnable;
.param p3, "delayError" # Z
.line 172
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
invoke-direct {p0}, Lio/reactivex/processors/FlowableProcessor;-><init>()V
.line 173
new-instance v0, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
const-string v1, "capacityHint"
invoke-static {p1, v1}, Lio/reactivex/internal/functions/ObjectHelper;->verifyPositive(ILjava/lang/String;)I
move-result v1
invoke-direct {v0, v1}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;-><init>(I)V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->queue:Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
.line 174
new-instance v0, Ljava/util/concurrent/atomic/AtomicReference;
invoke-direct {v0, p2}, Ljava/util/concurrent/atomic/AtomicReference;-><init>(Ljava/lang/Object;)V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->onTerminate:Ljava/util/concurrent/atomic/AtomicReference;
.line 175
iput-boolean p3, p0, Lio/reactivex/processors/UnicastProcessor;->delayError:Z
.line 176
new-instance v0, Ljava/util/concurrent/atomic/AtomicReference;
invoke-direct {v0}, Ljava/util/concurrent/atomic/AtomicReference;-><init>()V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
.line 177
new-instance v0, Ljava/util/concurrent/atomic/AtomicBoolean;
invoke-direct {v0}, Ljava/util/concurrent/atomic/AtomicBoolean;-><init>()V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->once:Ljava/util/concurrent/atomic/AtomicBoolean;
.line 178
new-instance v0, Lio/reactivex/processors/UnicastProcessor$UnicastQueueSubscription;
invoke-direct {v0, p0}, Lio/reactivex/processors/UnicastProcessor$UnicastQueueSubscription;-><init>(Lio/reactivex/processors/UnicastProcessor;)V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
.line 179
new-instance v0, Ljava/util/concurrent/atomic/AtomicLong;
invoke-direct {v0}, Ljava/util/concurrent/atomic/AtomicLong;-><init>()V
iput-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->requested:Ljava/util/concurrent/atomic/AtomicLong;
.line 180
return-void
.end method
.method public static create()Lio/reactivex/processors/UnicastProcessor;
.registers 2
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">()",
"Lio/reactivex/processors/UnicastProcessor<",
"TT;>;"
}
.end annotation
.annotation runtime Lio/reactivex/annotations/CheckReturnValue;
.end annotation
.line 78
new-instance v0, Lio/reactivex/processors/UnicastProcessor;
invoke-static {}, Lio/reactivex/processors/UnicastProcessor;->bufferSize()I
move-result v1
invoke-direct {v0, v1}, Lio/reactivex/processors/UnicastProcessor;-><init>(I)V
return-object v0
.end method
.method public static create(I)Lio/reactivex/processors/UnicastProcessor;
.registers 2
.param p0, "capacityHint" # I
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">(I)",
"Lio/reactivex/processors/UnicastProcessor<",
"TT;>;"
}
.end annotation
.annotation runtime Lio/reactivex/annotations/CheckReturnValue;
.end annotation
.line 89
new-instance v0, Lio/reactivex/processors/UnicastProcessor;
invoke-direct {v0, p0}, Lio/reactivex/processors/UnicastProcessor;-><init>(I)V
return-object v0
.end method
.method public static create(ILjava/lang/Runnable;)Lio/reactivex/processors/UnicastProcessor;
.registers 3
.param p0, "capacityHint" # I
.param p1, "onCancelled" # Ljava/lang/Runnable;
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">(I",
"Ljava/lang/Runnable;",
")",
"Lio/reactivex/processors/UnicastProcessor<",
"TT;>;"
}
.end annotation
.annotation runtime Lio/reactivex/annotations/CheckReturnValue;
.end annotation
.line 119
const-string v0, "onTerminate"
invoke-static {p1, v0}, Lio/reactivex/internal/functions/ObjectHelper;->requireNonNull(Ljava/lang/Object;Ljava/lang/String;)Ljava/lang/Object;
.line 120
new-instance v0, Lio/reactivex/processors/UnicastProcessor;
invoke-direct {v0, p0, p1}, Lio/reactivex/processors/UnicastProcessor;-><init>(ILjava/lang/Runnable;)V
return-object v0
.end method
.method public static create(ILjava/lang/Runnable;Z)Lio/reactivex/processors/UnicastProcessor;
.registers 4
.param p0, "capacityHint" # I
.param p1, "onCancelled" # Ljava/lang/Runnable;
.param p2, "delayError" # Z
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">(I",
"Ljava/lang/Runnable;",
"Z)",
"Lio/reactivex/processors/UnicastProcessor<",
"TT;>;"
}
.end annotation
.annotation runtime Lio/reactivex/annotations/CheckReturnValue;
.end annotation
.line 140
const-string v0, "onTerminate"
invoke-static {p1, v0}, Lio/reactivex/internal/functions/ObjectHelper;->requireNonNull(Ljava/lang/Object;Ljava/lang/String;)Ljava/lang/Object;
.line 141
new-instance v0, Lio/reactivex/processors/UnicastProcessor;
invoke-direct {v0, p0, p1, p2}, Lio/reactivex/processors/UnicastProcessor;-><init>(ILjava/lang/Runnable;Z)V
return-object v0
.end method
.method public static create(Z)Lio/reactivex/processors/UnicastProcessor;
.registers 4
.param p0, "delayError" # Z
.annotation system Ldalvik/annotation/Signature;
value = {
"<T:",
"Ljava/lang/Object;",
">(Z)",
"Lio/reactivex/processors/UnicastProcessor<",
"TT;>;"
}
.end annotation
.annotation runtime Lio/reactivex/annotations/CheckReturnValue;
.end annotation
.line 102
new-instance v0, Lio/reactivex/processors/UnicastProcessor;
invoke-static {}, Lio/reactivex/processors/UnicastProcessor;->bufferSize()I
move-result v1
const/4 v2, 0x0
invoke-direct {v0, v1, v2, p0}, Lio/reactivex/processors/UnicastProcessor;-><init>(ILjava/lang/Runnable;Z)V
return-object v0
.end method
# virtual methods
.method checkTerminated(ZZZLorg/reactivestreams/Subscriber;Lio/reactivex/internal/queue/SpscLinkedArrayQueue;)Z
.registers 10
.param p1, "failFast" # Z
.param p2, "d" # Z
.param p3, "empty" # Z
.annotation system Ldalvik/annotation/Signature;
value = {
"(ZZZ",
"Lorg/reactivestreams/Subscriber<",
"-TT;>;",
"Lio/reactivex/internal/queue/SpscLinkedArrayQueue<",
"TT;>;)Z"
}
.end annotation
.line 303
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
.local p4, "a":Lorg/reactivestreams/Subscriber;, "Lorg/reactivestreams/Subscriber<-TT;>;"
.local p5, "q":Lio/reactivex/internal/queue/SpscLinkedArrayQueue;, "Lio/reactivex/internal/queue/SpscLinkedArrayQueue<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
const/4 v1, 0x1
const/4 v2, 0x0
if-eqz v0, :cond_f
.line 304
invoke-virtual {p5}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->clear()V
.line 305
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v0, v2}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 306
return v1
.line 309
:cond_f
if-eqz p2, :cond_38
.line 310
if-eqz p1, :cond_25
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
if-eqz v0, :cond_25
.line 311
invoke-virtual {p5}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->clear()V
.line 312
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v0, v2}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 313
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
invoke-interface {p4, v0}, Lorg/reactivestreams/Subscriber;->onError(Ljava/lang/Throwable;)V
.line 314
return v1
.line 316
:cond_25
if-eqz p3, :cond_38
.line 317
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
.line 318
.local v0, "e":Ljava/lang/Throwable;
iget-object v3, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v3, v2}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 319
if-eqz v0, :cond_34
.line 320
invoke-interface {p4, v0}, Lorg/reactivestreams/Subscriber;->onError(Ljava/lang/Throwable;)V
goto :goto_37
.line 322
:cond_34
invoke-interface {p4}, Lorg/reactivestreams/Subscriber;->onComplete()V
.line 324
:goto_37
return v1
.line 328
.end local v0 # "e":Ljava/lang/Throwable;
:cond_38
const/4 v0, 0x0
return v0
.end method
.method doTerminate()V
.registers 4
.line 183
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->onTerminate:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v0}, Ljava/util/concurrent/atomic/AtomicReference;->get()Ljava/lang/Object;
move-result-object v0
check-cast v0, Ljava/lang/Runnable;
.line 184
.local v0, "r":Ljava/lang/Runnable;
if-eqz v0, :cond_16
iget-object v1, p0, Lio/reactivex/processors/UnicastProcessor;->onTerminate:Ljava/util/concurrent/atomic/AtomicReference;
const/4 v2, 0x0
invoke-virtual {v1, v0, v2}, Ljava/util/concurrent/atomic/AtomicReference;->compareAndSet(Ljava/lang/Object;Ljava/lang/Object;)Z
move-result v1
if-eqz v1, :cond_16
.line 185
invoke-interface {v0}, Ljava/lang/Runnable;->run()V
.line 187
:cond_16
return-void
.end method
.method drain()V
.registers 5
.line 276
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
invoke-virtual {v0}, Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;->getAndIncrement()I
move-result v0
if-eqz v0, :cond_9
.line 277
return-void
.line 280
:cond_9
const/4 v0, 0x1
.line 282
.local v0, "missed":I
iget-object v1, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v1}, Ljava/util/concurrent/atomic/AtomicReference;->get()Ljava/lang/Object;
move-result-object v1
check-cast v1, Lorg/reactivestreams/Subscriber;
.line 284
.local v1, "a":Lorg/reactivestreams/Subscriber;, "Lorg/reactivestreams/Subscriber<-TT;>;"
:goto_12
if-eqz v1, :cond_20
.line 286
iget-boolean v2, p0, Lio/reactivex/processors/UnicastProcessor;->enableOperatorFusion:Z
if-eqz v2, :cond_1c
.line 287
invoke-virtual {p0, v1}, Lio/reactivex/processors/UnicastProcessor;->drainFused(Lorg/reactivestreams/Subscriber;)V
goto :goto_1f
.line 289
:cond_1c
invoke-virtual {p0, v1}, Lio/reactivex/processors/UnicastProcessor;->drainRegular(Lorg/reactivestreams/Subscriber;)V
.line 291
:goto_1f
return-void
.line 294
:cond_20
iget-object v2, p0, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
neg-int v3, v0
invoke-virtual {v2, v3}, Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;->addAndGet(I)I
move-result v0
.line 295
if-nez v0, :cond_2b
.line 296
nop
.line 300
return-void
.line 298
:cond_2b
iget-object v2, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v2}, Ljava/util/concurrent/atomic/AtomicReference;->get()Ljava/lang/Object;
move-result-object v2
move-object v1, v2
check-cast v1, Lorg/reactivestreams/Subscriber;
goto :goto_12
.end method
.method drainFused(Lorg/reactivestreams/Subscriber;)V
.registers 8
.annotation system Ldalvik/annotation/Signature;
value = {
"(",
"Lorg/reactivestreams/Subscriber<",
"-TT;>;)V"
}
.end annotation
.line 234
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
.local p1, "a":Lorg/reactivestreams/Subscriber;, "Lorg/reactivestreams/Subscriber<-TT;>;"
const/4 v0, 0x1
.line 236
.local v0, "missed":I
iget-object v1, p0, Lio/reactivex/processors/UnicastProcessor;->queue:Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
.line 237
.local v1, "q":Lio/reactivex/internal/queue/SpscLinkedArrayQueue;, "Lio/reactivex/internal/queue/SpscLinkedArrayQueue<TT;>;"
iget-boolean v2, p0, Lio/reactivex/processors/UnicastProcessor;->delayError:Z
xor-int/lit8 v2, v2, 0x1
.line 240
.local v2, "failFast":Z
:goto_7
iget-boolean v3, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
const/4 v4, 0x0
if-eqz v3, :cond_15
.line 241
invoke-virtual {v1}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->clear()V
.line 242
iget-object v3, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v3, v4}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 243
return-void
.line 246
:cond_15
iget-boolean v3, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
.line 248
.local v3, "d":Z
if-eqz v2, :cond_2d
if-eqz v3, :cond_2d
iget-object v5, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
if-eqz v5, :cond_2d
.line 249
invoke-virtual {v1}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->clear()V
.line 250
iget-object v5, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v5, v4}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 251
iget-object v4, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
invoke-interface {p1, v4}, Lorg/reactivestreams/Subscriber;->onError(Ljava/lang/Throwable;)V
.line 252
return-void
.line 254
:cond_2d
invoke-interface {p1, v4}, Lorg/reactivestreams/Subscriber;->onNext(Ljava/lang/Object;)V
.line 256
if-eqz v3, :cond_43
.line 257
iget-object v5, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v5, v4}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
.line 259
iget-object v4, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
.line 260
.local v4, "ex":Ljava/lang/Throwable;
if-eqz v4, :cond_3f
.line 261
invoke-interface {p1, v4}, Lorg/reactivestreams/Subscriber;->onError(Ljava/lang/Throwable;)V
goto :goto_42
.line 263
:cond_3f
invoke-interface {p1}, Lorg/reactivestreams/Subscriber;->onComplete()V
.line 265
:goto_42
return-void
.line 268
.end local v4 # "ex":Ljava/lang/Throwable;
:cond_43
iget-object v4, p0, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
neg-int v5, v0
invoke-virtual {v4, v5}, Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;->addAndGet(I)I
move-result v0
.line 269
if-nez v0, :cond_4e
.line 270
nop
.line 273
.end local v3 # "d":Z
return-void
.line 272
:cond_4e
goto :goto_7
.end method
.method drainRegular(Lorg/reactivestreams/Subscriber;)V
.registers 19
.annotation system Ldalvik/annotation/Signature;
value = {
"(",
"Lorg/reactivestreams/Subscriber<",
"-TT;>;)V"
}
.end annotation
.line 190
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
.local p1, "a":Lorg/reactivestreams/Subscriber;, "Lorg/reactivestreams/Subscriber<-TT;>;"
move-object/from16 v8, p0
const/4 v0, 0x1
.line 192
.local v0, "missed":I
iget-object v9, v8, Lio/reactivex/processors/UnicastProcessor;->queue:Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
.line 193
.local v9, "q":Lio/reactivex/internal/queue/SpscLinkedArrayQueue;, "Lio/reactivex/internal/queue/SpscLinkedArrayQueue<TT;>;"
iget-boolean v1, v8, Lio/reactivex/processors/UnicastProcessor;->delayError:Z
const/4 v10, 0x1
xor-int/2addr v1, v10
move v11, v0
.line 196
.end local v0 # "missed":I
.local v1, "failFast":Z
.local v11, "missed":I
:goto_a
iget-object v0, v8, Lio/reactivex/processors/UnicastProcessor;->requested:Ljava/util/concurrent/atomic/AtomicLong;
invoke-virtual {v0}, Ljava/util/concurrent/atomic/AtomicLong;->get()J
move-result-wide v12
.line 197
.local v12, "r":J
const-wide/16 v2, 0x0
move-wide v14, v2
.line 199
.local v14, "e":J
:goto_13
cmp-long v0, v12, v14
if-eqz v0, :cond_41
.line 200
iget-boolean v6, v8, Lio/reactivex/processors/UnicastProcessor;->done:Z
.line 202
.local v6, "d":Z
invoke-virtual {v9}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->poll()Ljava/lang/Object;
move-result-object v7
.line 203
.local v7, "t":Ljava/lang/Object;, "TT;"
if-nez v7, :cond_21
const/4 v0, 0x1
goto :goto_22
:cond_21
const/4 v0, 0x0
:goto_22
move/from16 v16, v0
.line 205
.local v16, "empty":Z
move-object/from16 v0, p0
move v2, v6
move/from16 v3, v16
move-object/from16 v4, p1
move-object v5, v9
invoke-virtual/range {v0 .. v5}, Lio/reactivex/processors/UnicastProcessor;->checkTerminated(ZZZLorg/reactivestreams/Subscriber;Lio/reactivex/internal/queue/SpscLinkedArrayQueue;)Z
move-result v0
if-eqz v0, :cond_33
.line 206
return-void
.line 209
:cond_33
if-eqz v16, :cond_38
.line 210
move-object/from16 v0, p1
goto :goto_43
.line 213
:cond_38
move-object/from16 v0, p1
invoke-interface {v0, v7}, Lorg/reactivestreams/Subscriber;->onNext(Ljava/lang/Object;)V
.line 215
const-wide/16 v2, 0x1
add-long/2addr v14, v2
.line 216
.end local v6 # "d":Z
.end local v7 # "t":Ljava/lang/Object;, "TT;"
.end local v16 # "empty":Z
goto :goto_13
.line 199
:cond_41
move-object/from16 v0, p1
.line 218
:goto_43
cmp-long v2, v12, v14
if-nez v2, :cond_5a
iget-boolean v4, v8, Lio/reactivex/processors/UnicastProcessor;->done:Z
invoke-virtual {v9}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->isEmpty()Z
move-result v5
move-object/from16 v2, p0
move v3, v1
move-object/from16 v6, p1
move-object v7, v9
invoke-virtual/range {v2 .. v7}, Lio/reactivex/processors/UnicastProcessor;->checkTerminated(ZZZLorg/reactivestreams/Subscriber;Lio/reactivex/internal/queue/SpscLinkedArrayQueue;)Z
move-result v2
if-eqz v2, :cond_5a
.line 219
return-void
.line 222
:cond_5a
const-wide/16 v2, 0x0
cmp-long v4, v14, v2
if-eqz v4, :cond_6f
const-wide v2, 0x7fffffffffffffffL
cmp-long v4, v12, v2
if-eqz v4, :cond_6f
.line 223
iget-object v2, v8, Lio/reactivex/processors/UnicastProcessor;->requested:Ljava/util/concurrent/atomic/AtomicLong;
neg-long v3, v14
invoke-virtual {v2, v3, v4}, Ljava/util/concurrent/atomic/AtomicLong;->addAndGet(J)J
.line 226
:cond_6f
iget-object v2, v8, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
neg-int v3, v11
invoke-virtual {v2, v3}, Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;->addAndGet(I)I
move-result v11
.line 227
if-nez v11, :cond_7a
.line 228
nop
.line 231
.end local v12 # "r":J
.end local v14 # "e":J
return-void
.line 230
:cond_7a
goto :goto_a
.end method
.method public getThrowable()Ljava/lang/Throwable;
.registers 2
.line 466
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-eqz v0, :cond_7
.line 467
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
return-object v0
.line 469
:cond_7
const/4 v0, 0x0
return-object v0
.end method
.method public hasComplete()Z
.registers 2
.line 474
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-eqz v0, :cond_a
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
if-nez v0, :cond_a
const/4 v0, 0x1
goto :goto_b
:cond_a
const/4 v0, 0x0
:goto_b
return v0
.end method
.method public hasSubscribers()Z
.registers 2
.line 461
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v0}, Ljava/util/concurrent/atomic/AtomicReference;->get()Ljava/lang/Object;
move-result-object v0
if-eqz v0, :cond_a
const/4 v0, 0x1
goto :goto_b
:cond_a
const/4 v0, 0x0
:goto_b
return v0
.end method
.method public hasThrowable()Z
.registers 2
.line 479
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-eqz v0, :cond_a
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
if-eqz v0, :cond_a
const/4 v0, 0x1
goto :goto_b
:cond_a
const/4 v0, 0x0
:goto_b
return v0
.end method
.method public onComplete()V
.registers 2
.line 376
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-nez v0, :cond_13
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
if-eqz v0, :cond_9
goto :goto_13
.line 380
:cond_9
const/4 v0, 0x1
iput-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
.line 382
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->doTerminate()V
.line 384
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->drain()V
.line 385
return-void
.line 377
:cond_13
:goto_13
return-void
.end method
.method public onError(Ljava/lang/Throwable;)V
.registers 4
.param p1, "t" # Ljava/lang/Throwable;
.line 357
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-nez v0, :cond_1f
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
if-eqz v0, :cond_9
goto :goto_1f
.line 362
:cond_9
if-nez p1, :cond_13
.line 363
new-instance v0, Ljava/lang/NullPointerException;
const-string v1, "onError called with null. Null values are generally not allowed in 2.x operators and sources."
invoke-direct {v0, v1}, Ljava/lang/NullPointerException;-><init>(Ljava/lang/String;)V
move-object p1, v0
.line 366
:cond_13
iput-object p1, p0, Lio/reactivex/processors/UnicastProcessor;->error:Ljava/lang/Throwable;
.line 367
const/4 v0, 0x1
iput-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
.line 369
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->doTerminate()V
.line 371
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->drain()V
.line 372
return-void
.line 358
:cond_1f
:goto_1f
invoke-static {p1}, Lio/reactivex/plugins/RxJavaPlugins;->onError(Ljava/lang/Throwable;)V
.line 359
return-void
.end method
.method public onNext(Ljava/lang/Object;)V
.registers 4
.annotation system Ldalvik/annotation/Signature;
value = {
"(TT;)V"
}
.end annotation
.line 342
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
.local p1, "t":Ljava/lang/Object;, "TT;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-nez v0, :cond_1f
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
if-eqz v0, :cond_9
goto :goto_1f
.line 346
:cond_9
if-nez p1, :cond_16
.line 347
new-instance v0, Ljava/lang/NullPointerException;
const-string v1, "onNext called with null. Null values are generally not allowed in 2.x operators and sources."
invoke-direct {v0, v1}, Ljava/lang/NullPointerException;-><init>(Ljava/lang/String;)V
invoke-virtual {p0, v0}, Lio/reactivex/processors/UnicastProcessor;->onError(Ljava/lang/Throwable;)V
.line 348
return-void
.line 351
:cond_16
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->queue:Lio/reactivex/internal/queue/SpscLinkedArrayQueue;
invoke-virtual {v0, p1}, Lio/reactivex/internal/queue/SpscLinkedArrayQueue;->offer(Ljava/lang/Object;)Z
.line 352
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->drain()V
.line 353
return-void
.line 343
:cond_1f
:goto_1f
return-void
.end method
.method public onSubscribe(Lorg/reactivestreams/Subscription;)V
.registers 4
.param p1, "s" # Lorg/reactivestreams/Subscription;
.line 333
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->done:Z
if-nez v0, :cond_12
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
if-eqz v0, :cond_9
goto :goto_12
.line 336
:cond_9
const-wide v0, 0x7fffffffffffffffL
invoke-interface {p1, v0, v1}, Lorg/reactivestreams/Subscription;->request(J)V
goto :goto_15
.line 334
:cond_12
:goto_12
invoke-interface {p1}, Lorg/reactivestreams/Subscription;->cancel()V
.line 338
:goto_15
return-void
.end method
.method protected subscribeActual(Lorg/reactivestreams/Subscriber;)V
.registers 5
.annotation system Ldalvik/annotation/Signature;
value = {
"(",
"Lorg/reactivestreams/Subscriber<",
"-TT;>;)V"
}
.end annotation
.line 389
.local p0, "this":Lio/reactivex/processors/UnicastProcessor;, "Lio/reactivex/processors/UnicastProcessor<TT;>;"
.local p1, "s":Lorg/reactivestreams/Subscriber;, "Lorg/reactivestreams/Subscriber<-TT;>;"
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->once:Ljava/util/concurrent/atomic/AtomicBoolean;
invoke-virtual {v0}, Ljava/util/concurrent/atomic/AtomicBoolean;->get()Z
move-result v0
if-nez v0, :cond_2b
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->once:Ljava/util/concurrent/atomic/AtomicBoolean;
const/4 v1, 0x0
const/4 v2, 0x1
invoke-virtual {v0, v1, v2}, Ljava/util/concurrent/atomic/AtomicBoolean;->compareAndSet(ZZ)Z
move-result v0
if-eqz v0, :cond_2b
.line 391
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->wip:Lio/reactivex/internal/subscriptions/BasicIntQueueSubscription;
invoke-interface {p1, v0}, Lorg/reactivestreams/Subscriber;->onSubscribe(Lorg/reactivestreams/Subscription;)V
.line 392
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
invoke-virtual {v0, p1}, Ljava/util/concurrent/atomic/AtomicReference;->set(Ljava/lang/Object;)V
.line 393
iget-boolean v0, p0, Lio/reactivex/processors/UnicastProcessor;->cancelled:Z
if-eqz v0, :cond_27
.line 394
iget-object v0, p0, Lio/reactivex/processors/UnicastProcessor;->actual:Ljava/util/concurrent/atomic/AtomicReference;
const/4 v1, 0x0
invoke-virtual {v0, v1}, Ljava/util/concurrent/atomic/AtomicReference;->lazySet(Ljava/lang/Object;)V
goto :goto_35
.line 396
:cond_27
invoke-virtual {p0}, Lio/reactivex/processors/UnicastProcessor;->drain()V
goto :goto_35
.line 399
:cond_2b
new-instance v0, Ljava/lang/IllegalStateException;
const-string v1, "This processor allows only a single Subscriber"
invoke-direct {v0, v1}, Ljava/lang/IllegalStateException;-><init>(Ljava/lang/String;)V
invoke-static {v0, p1}, Lio/reactivex/internal/subscriptions/EmptySubscription;->error(Ljava/lang/Throwable;Lorg/reactivestreams/Subscriber;)V
.line 401
:goto_35
return-void
.end method