package rsc.publisher;

import java.util.Objects;

import org.reactivestreams.Subscriber;

import rsc.flow.Cancellation;
import rsc.scheduler.Scheduler;
import rsc.util.EmptySubscription;
import rsc.util.ExceptionHelper;

/**
 * Publisher indicating a scalar/empty source that subscribes on the specified scheduler.
 * 
 * @param <T>
 */
final class PublisherSubscribeOnValue<T> extends Px<T> {

    final T value;
    
    final Scheduler scheduler;

    public PublisherSubscribeOnValue(T value, 
            Scheduler scheduler) {
        this.value = value;
        this.scheduler = Objects.requireNonNull(scheduler, "scheduler");
    }

    @Override
    public void subscribe(Subscriber<? super T> s) {
        Scheduler.Worker worker;
        
        try {
            worker = scheduler.createWorker();
        } catch (Throwable e) {
            ExceptionHelper.throwIfFatal(e);
            EmptySubscription.error(s, e);
            return;
        }
        
        if (worker == null) {
            EmptySubscription.error(s, new NullPointerException("The scheduler returned a null Function"));
            return;
        }

        T v = value;
        if (v == null) {
            PublisherSubscribeOn.ScheduledEmpty parent = new PublisherSubscribeOn.ScheduledEmpty(s);
            s.onSubscribe(parent);
            Cancellation f = scheduler.schedule(parent);
            parent.setFuture(f);
        } else {
            s.onSubscribe(new PublisherSubscribeOn.ScheduledScalar<>(s, v, scheduler));
        }
    }
}

@@ -149,76 +149,8 @@
 ler;
-%0Aimport rsc.util.EmptySubscription;%0Aimport rsc.util.ExceptionHelper;
 %0A%0A/*
@@ -638,439 +638,8 @@
 ) %7B%0A
-        Scheduler.Worker worker;%0A        %0A        try %7B%0A            worker = scheduler.createWorker();%0A        %7D catch (Throwable e) %7B%0A            ExceptionHelper.throwIfFatal(e);%0A            EmptySubscription.error(s, e);%0A            return;%0A        %7D%0A        %0A        if (worker == null) %7B%0A            EmptySubscription.error(s, new NullPointerException(%22The scheduler returned a null Function%22));%0A            return;%0A        %7D%0A%0A
     

Remove unused worker (and thus a worker leak)