Java 类io.reactivex.processors.BehaviorProcessor 实例源码

@Test
public void testBasicTransform()
{
    BehaviorProcessor<List<Integer>> processor = BehaviorProcessor.create();
    FlowableList<Integer> list = FlowableList.diff(processor);
    TestSubscriber<Update<Integer>> test = list.updates().test();

    processor.onNext(Arrays.asList(1, 2, 3, 4));

    Update<Integer> firstUpdate = test.values().get(0);
    assertEquals(Collections.singletonList(Change.reloaded()), firstUpdate.changes);

    processor.onNext(Arrays.asList(2, 4, 5));

    Update<Integer> secondUpdate = test.values().get(1);

    assertEquals(Arrays.asList(2, 4, 5), secondUpdate.list);
    assertEquals(Arrays.asList(
            2, 4, 5),
            TestTools.applyChanges(firstUpdate.list, secondUpdate.list, secondUpdate.changes));
}

@SafeVarargs
public ReduxFXStore(S initialState, BiFunction<S, Object, Update<S>> updater, Middleware<S>... middlewares) {
    final BiFunction<S, Object, Update<S>> chainedUpdater = applyMiddlewares(updater, middlewares);

    final Publisher<Object> actionPublisher =
            Flowable.create(actionEmitter -> this.actionEmitter = actionEmitter, BackpressureStrategy.BUFFER);

    final FlowableProcessor<Update<S>> updateProcessor = BehaviorProcessor.create();

    statePublisher = updateProcessor.map(Update::getState)
            .startWith(initialState);

    statePublisher.zipWith(actionPublisher, chainedUpdater::apply)
            .subscribe(updateProcessor);

    commandPublisher = updateProcessor
            .map(Update::getCommands)
            .flatMapIterable(commands -> commands);
}

public static FlowableProcessor<DownloadEvent> createProcessor(
        String missionId, Map<String, FlowableProcessor<DownloadEvent>> processorMap) {

    if (processorMap.get(missionId) == null) {
        FlowableProcessor<DownloadEvent> processor =
                BehaviorProcessor.<DownloadEvent>create().toSerialized();
        processorMap.put(missionId, processor);
    }
    return processorMap.get(missionId);
}

/**
 * Constructs a new, empty SubjectMap
 */
public SubjectMap()
{
    ReadWriteLock _readWriteLock = new ReentrantReadWriteLock();

    _readLock = _readWriteLock.readLock();
    _writeLock = _readWriteLock.writeLock();

    _weakCache = new HashMap<>();
    _cache = new HashMap<>();
    _faults = BehaviorProcessor.create();

    _weakSources = new HashMap<>();
}

private Processor<V, V> attachSource(K key)
{
    _writeLock.lock();
    try {
        // if our source is being attached, we expect that all existing sources have been
        // cleaned up properly. If not, this is a serious issue
        assert(!_weakSources.containsKey(key));

        Processor<V, V> value = BehaviorProcessor.create();

        WeakReference<Flowable<V>> weakConnector = _weakCache.get(key);

        // if an observable is being attached then it must have been added to the weak cache
        // and it must still be referenced
        Flowable<V> connector = weakConnector.get();

        // the observable must be retained by someone since it is being attached
        assert(connector != null);

        // strongly retain the observable and add the subject so future next
        // calls will be piped through the subject
        _weakSources.put(key, new WeakReference<>(value));
        _cache.put(key, connector);

        return value;
    }
    finally {
        _writeLock.unlock();
    }
}

@Test
public void testSortedMoveOnly()
{
    final List<String> list1 = Arrays.asList("C", "B", "J", "D", "G", "H", "A", "I", "E", "F");
    final List<String> list2 = Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J");

    BehaviorProcessor<List<String>> processor = BehaviorProcessor.create();
    FlowableList<String> list = FlowableList.diff(processor);
    TestSubscriber<Update<String>> test = list.updates().test();

    processor.onNext(list1);

    Update<String> firstUpdate = test.values().get(0);
    assertEquals(Collections.singletonList(Change.reloaded()), firstUpdate.changes);

    processor.onNext(list2);

    Update<String> secondUpdate = test.values().get(1);

    assertEquals(
            Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J"),
            secondUpdate.list);

    assertEquals(
            Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J"),
            TestTools.applyChanges(firstUpdate.list, secondUpdate.list, secondUpdate.changes));

    for (Change change : secondUpdate.changes) {
        assertEquals(change.type, Change.Type.Moved);
    }
}

@Test
public void testSortedIgnoreMoves()
{
    final List<String> list1 = Arrays.asList("C", "B", "J", "D", "G", "H", "A", "I", "E", "F");
    final List<String> list2 = Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J");

    BehaviorProcessor<List<String>> processor = BehaviorProcessor.create();
    FlowableList<String> list = FlowableList.diff(processor, false);
    TestSubscriber<Update<String>> test = list.updates().test();

    processor.onNext(list1);

    Update<String> firstUpdate = test.values().get(0);
    assertEquals(Collections.singletonList(Change.reloaded()), firstUpdate.changes);

    processor.onNext(list2);

    Update<String> secondUpdate = test.values().get(1);

    assertEquals(
            Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J"),
            secondUpdate.list);

    assertEquals(
            Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I", "J"),
            TestTools.applyChanges(firstUpdate.list, secondUpdate.list, secondUpdate.changes));

    for (Change change : secondUpdate.changes) {
        assertNotEquals(change.type, Change.Type.Moved);
    }
}

private ObservableChoosable(Iterable<T> allValues, T defaultValue) {
    requireNonNull(defaultValue, "the default value must not be null");
    requireNonNull(allValues, "allValues must not be null");
    this.actualValueSubject = BehaviorProcessor.createDefault(defaultValue);
    this.allValuesSubject = BehaviorProcessor.createDefault(allValues);
}

public RunStateServiceImpl(RunState initialState) {
    runState = BehaviorProcessor.createDefault(requireNonNull(initialState, "initial state must not be null"));
}

public GCounter(String nodeId, String crdtId) {
    super(nodeId, crdtId, BehaviorProcessor.create());
}

public PNCounter(String nodeId, String crtdId) {
    super(nodeId, crtdId, BehaviorProcessor.create());
}

public LWWRegister(String nodeId, String crdtId) {
    super(nodeId, crdtId, BehaviorProcessor.create());
    this.clock = new StrictVectorClock(nodeId);
}

public RequestBodyWrapper(@NonNull RequestBody requestBody, String filePath) {
    this.mRequestBody = requestBody;
    this.mFilePath = filePath;
    this.mUploadProcessor = BehaviorProcessor.create();
}

public static RxJava2ProcProxy behaviorProcessorProxy() {
    return new RxJava2ProcProxy(BehaviorProcessor.create(), Roxy.TePolicy.PASS);
}

public static RxJava2ProcProxy serializedBehaviorProcessorProxy() {
    return new RxJava2ProcProxy(BehaviorProcessor.create().toSerialized(), Roxy.TePolicy.PASS);
}

public static RxJava2ProcProxy safeBehaviorProcessorProxy() {
    return new RxJava2ProcProxy(BehaviorProcessor.create(), Roxy.TePolicy.WRAP);
}

public static RxJava2ProcProxy safeSerializedBehaviorProcessorProxy() {
    return new RxJava2ProcProxy(BehaviorProcessor.create().toSerialized(), Roxy.TePolicy.WRAP);
}

@Test
public void testAndShortCircuiting()
{
    BehaviorProcessor<Boolean> bool1 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool2 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool3 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool4 = BehaviorProcessor.create();

    bool1.onNext(true);
    bool2.onNext(true);
    bool3.onNext(true);
    bool4.onNext(true);

    Flowable<Boolean> joined = BooleanFlowables.and(bool1, bool2, bool3, bool4);

    TestSubscriber<Boolean> results = new TestSubscriber<>();

    joined.subscribe(results);

    results.assertValues(true);

    bool2.onNext(false);

    results.assertValues(true, false);

    bool3.onNext(false);

    results.assertValues(true, false);

    bool3.onNext(true);

    results.assertValues(true, false);

    bool2.onNext(true);

    results.assertValues(true, false, true);

    bool1.onNext(false);

    results.assertValues(true, false, true, false);

    bool4.onNext(false);

    results.assertValues(true, false, true, false);
}

@Test
public void testOrShortCircuiting()
{
    BehaviorProcessor<Boolean> bool1 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool2 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool3 = BehaviorProcessor.create();
    BehaviorProcessor<Boolean> bool4 = BehaviorProcessor.create();

    bool1.onNext(true);
    bool2.onNext(false);
    bool3.onNext(false);
    bool4.onNext(false);

    Flowable<Boolean> joined = BooleanFlowables.or(bool1, bool2, bool3, bool4);

    TestSubscriber<Boolean> results = new TestSubscriber<>();

    joined.subscribe(results);

    results.assertValues(true);

    bool2.onNext(true);

    results.assertValues(true);

    bool1.onNext(false);

    results.assertValues(true);

    bool2.onNext(false);

    results.assertValues(true, false);

    bool3.onNext(true);

    results.assertValues(true, false, true);

    bool4.onNext(true);

    results.assertValues(true, false, true);

    bool4.onNext(false);

    results.assertValues(true, false, true);
}