OnSubscribeJoin.java 文件源码

@Override
public void onNext(TRight args) {
    int id; 
    int highLeftId;
    synchronized (guard) {
        id = rightId++;
        rightMap.put(id, args);
        highLeftId = leftId;
    }
    SerialSubscription md = new SerialSubscription();
    group.add(md);

    Observable<TRightDuration> duration;
    try {
        duration = rightDurationSelector.call(args);

        Subscriber<TRightDuration> d2 = new RightDurationSubscriber(id);
        group.add(d2);

        duration.unsafeSubscribe(d2);


        List<TLeft> leftValues = new ArrayList<TLeft>();
        synchronized (guard) {
            for (Map.Entry<Integer, TLeft> entry : leftMap.entrySet()) {
                if (entry.getKey() < highLeftId) {
                    leftValues.add(entry.getValue());
                }
            }
        }

        for (TLeft lv : leftValues) {
            R result = resultSelector.call(lv, args);
            subscriber.onNext(result);
        }

    } catch (Throwable t) {
        onError(t);
    }
}