<commit_before>package com.github.davidmoten.rx2.internal.flowable;

import java.util.concurrent.atomic.AtomicInteger;

import org.reactivestreams.Subscriber;

import io.reactivex.Flowable;
import io.reactivex.flowables.ConnectableFlowable;
import io.reactivex.internal.fuseable.SimplePlainQueue;
import io.reactivex.internal.queue.MpscLinkedQueue;

public class FlowableRefCountNonBlocking<T> extends Flowable<T> {

    private final ConnectableFlowable<T> source;
    private final SimplePlainQueue<Object> queue;
    private final AtomicInteger wip = new AtomicInteger();

    public FlowableRefCountNonBlocking(ConnectableFlowable<T> source) {
        this.source = source;
        this.queue = new MpscLinkedQueue<Object>();
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {
        queue.offer(child);
        drain();
    }

    private void drain() {
        if (wip.getAndIncrement() == 0) {
            while (true) {
                Object o;
                while ((o = queue.poll()) != null) {
                    // TODO do something with o
                    if (wip.decrementAndGet() == 0) {
                        return;
                    }
                }
            }
        }
    }
}<commit_msg>ref count play<commit_after>package com.github.davidmoten.rx2.internal.flowable;

import java.util.concurrent.atomic.AtomicInteger;

import org.reactivestreams.Subscriber;

import io.reactivex.Flowable;
import io.reactivex.flowables.ConnectableFlowable;
import io.reactivex.internal.fuseable.SimplePlainQueue;
import io.reactivex.internal.queue.MpscLinkedQueue;

public class FlowableRefCountNonBlocking<T> extends Flowable<T> {

    private final ConnectableFlowable<T> source;
    private final SimplePlainQueue<Object> queue;
    private final AtomicInteger wip = new AtomicInteger();

    public FlowableRefCountNonBlocking(ConnectableFlowable<T> source) {
        this.source = source;
        this.queue = new MpscLinkedQueue<Object>();
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {
        queue.offer(child);
        drain();
    }
    
    private void drain() {
        if (wip.getAndIncrement() == 0) {
            while (true) {
                Object o;
                while ((o = queue.poll()) != null) {
                    // TODO do something with o
                    if (wip.decrementAndGet() == 0) {
                        return;
                    }
                }
            }
        }
    }
    
}
{'fraction_non_alphanumeric': 0.0971563981042654, 'fraction_numerical': 0.002369668246445498, 'mean_word_length': 2.2810880829015545, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5940593', 'n_tokens_mistral': 752, 'n_tokens_neox': 709, 'n_words': 148}