Java 类rx.functions.Action3 实例源码

private void initArticles(List<ZhihuStoryItem> items) {
    for (ZhihuStoryItem item : items)
        getAdapter().add(new ZhihuStoryItemViewModel(item)
                .onAction(new Action3<String, Integer, Integer>() {
                    @Override
                    public void call(String id, Integer touchX, Integer touchY) {
                        Intent intent = ZhihuStoryDetailActivity.intentFor(getContext());
                        intent.putExtra(Constant.EXTRA_ZHIHU_STORY_ID, id);
                        intent.putExtra(Constant.EXTRA_TOUCH_X, touchX);
                        intent.putExtra(Constant.EXTRA_TOUCH_Y, touchY);
                        getContext().startActivity(intent);
                    }
                }));
    getAdapter().add(new BlankViewModel(R.dimen.common_gap));
    getAdapter().notifyDataSetChanged();

}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @param a5 Action
 * @param a6 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4, @NonNull final Action3<? super T1, ? super T2, ? super T3> a5, @NonNull final Action3<? super T1, ? super T2, ? super T3> a6) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
            a5.call(t1, t2, t3);
            a6.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @param a5 Action
 * @param a6 Action
 * @param a7 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4, @NonNull final Action3<? super T1, ? super T2, ? super T3> a5, @NonNull final Action3<? super T1, ? super T2, ? super T3> a6, @NonNull final Action3<? super T1, ? super T2, ? super T3> a7) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
            a5.call(t1, t2, t3);
            a6.call(t1, t2, t3);
            a7.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @param a5 Action
 * @param a6 Action
 * @param a7 Action
 * @param a8 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4, @NonNull final Action3<? super T1, ? super T2, ? super T3> a5, @NonNull final Action3<? super T1, ? super T2, ? super T3> a6, @NonNull final Action3<? super T1, ? super T2, ? super T3> a7, @NonNull final Action3<? super T1, ? super T2, ? super T3> a8) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
            a5.call(t1, t2, t3);
            a6.call(t1, t2, t3);
            a7.call(t1, t2, t3);
            a8.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @param a5 Action
 * @param a6 Action
 * @param a7 Action
 * @param a8 Action
 * @param a9 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4, @NonNull final Action3<? super T1, ? super T2, ? super T3> a5, @NonNull final Action3<? super T1, ? super T2, ? super T3> a6, @NonNull final Action3<? super T1, ? super T2, ? super T3> a7, @NonNull final Action3<? super T1, ? super T2, ? super T3> a8, @NonNull final Action3<? super T1, ? super T2, ? super T3> a9) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
            a5.call(t1, t2, t3);
            a6.call(t1, t2, t3);
            a7.call(t1, t2, t3);
            a8.call(t1, t2, t3);
            a9.call(t1, t2, t3);
        }
    };
}

@Experimental
public static <S, T> Observable$OnSubscribe<T> createSingleState(Func0<? extends S> generator, final Action3<? super S, Long, ? super Observer<Observable<? extends T>>> next) {
    return new AsyncOnSubscribeImpl((Func0) generator, new Func3<S, Long, Observer<Observable<? extends T>>, S>() {
        public S call(S state, Long requested, Observer<Observable<? extends T>> subscriber) {
            next.call(state, requested, subscriber);
            return state;
        }
    });
}

@Experimental
public static <S, T> Observable$OnSubscribe<T> createSingleState(Func0<? extends S> generator, final Action3<? super S, Long, ? super Observer<Observable<? extends T>>> next, Action1<? super S> onUnsubscribe) {
    return new AsyncOnSubscribeImpl(generator, new Func3<S, Long, Observer<Observable<? extends T>>, S>() {
        public S call(S state, Long requested, Observer<Observable<? extends T>> subscriber) {
            next.call(state, requested, subscriber);
            return state;
        }
    }, onUnsubscribe);
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @param a2 Action
 * @param a3 Action
 * @param a4 Action
 * @param a5 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3> a1, @NonNull final Action3<? super T1, ? super T2, ? super T3> a2, @NonNull final Action3<? super T1, ? super T2, ? super T3> a3, @NonNull final Action3<? super T1, ? super T2, ? super T3> a4, @NonNull final Action3<? super T1, ? super T2, ? super T3> a5) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            a1.call(t1, t2, t3);
            a2.call(t1, t2, t3);
            a3.call(t1, t2, t3);
            a4.call(t1, t2, t3);
            a5.call(t1, t2, t3);
        }
    };
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a Actions
 * @return Single action stream combined from provided actions
 */
@SafeVarargs
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull final Action3<? super T1, ? super T2, ? super T3>... a) {
    return new Action3<T1, T2, T3>() {
        @Override
        public void call(T1 t1, T2 t2, T3 t3) {
            //noinspection ForLoopReplaceableByForEach
            for (int i = 0, count = a.length; i < count; i++) {
                a[i].call(t1, t2, t3);
            }
        }
    };
}

public void onBindViewHolder(VH viewHolder, int position, T item) { // final, DO NOT Override until certainly
    onBindViewHolderSupered = true;
    if (mOnBindViewHolder == null) {
        mOnBindViewHolder = new Action3<VH, Integer, T>() {
            @Override
            public void call(VH vh, Integer i, T t) {
                vh.onBind(i, t);
            }
        };
    }
    mOnBindViewHolder.call(viewHolder, position, item);
}

public MovieCollectItemViewModel(DoubanMovieItem item, Action1<DoubanMovieItem> onClick, Action3<String, String, Integer> onUncollect) {
    this.item = item;
    this.onClick = onClick;
    this.onUncollect = onUncollect;
}

public OneCommonCollectItemViewModel(String id, String imgUrl, String title, Action3<String, String, Integer> onUnCollect) {
    this.id = id;
    this.imgUrl = imgUrl;
    this.title = title;
    this.onUnCollect = onUnCollect;
}

public OneCommonCollectItemViewModel(Review review, Action3<String, String, Integer> onUnCollect) {
    this.review = review;
    this.onUnCollect = onUnCollect;
}

public StoryCollectItemViewModel(ZhihuStoryItem item, Action1<ZhihuStoryItem> onClick, Action3<String, String, Integer> onUncollect) {
    this.item = item;
    this.onClick = onClick;
    this.onUncollect = onUncollect;
}

private void initActions() {
    onMovieItemClick = new Action1<DoubanMovieItem>() {
        @Override
        public void call(DoubanMovieItem item) {
            Intent intent = DoubanMovieDetailActivity.intentFor(getContext());
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_ID, item.id);
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_POSTER, item.images.large);
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_TITLE, item.title);
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_MAINLAND_PUBDATE, item.mainlandPubdate);
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_RATING, item.rating.average + "/" + 10);
            getContext().startActivity(intent);
        }
    };
    onStoryItemClick = new Action1<ZhihuStoryItem>() {
        @Override
        public void call(ZhihuStoryItem zhihuStoryItem) {
            Intent intent = ZhihuStoryDetailActivity.intentFor(getContext());
            intent.putExtra(Constant.EXTRA_ZHIHU_STORY_ID, zhihuStoryItem.id);
            getContext().startActivity(intent);
        }
    };
    onUnCollect = new Action3<String, String, Integer>() {
        @Override
        public void call(final String id, final String type, final Integer position) {
            RxTask.asyncTask(new Action0() {
                @Override
                public void call() {
                    CollectionManager.getInstance(getContext()).remove(id, type);
                }
            }, new Action0() {
                @Override
                public void call() {
                    getAdapter().remove(position.intValue());
                    getAdapter().notifyItemRemoved(position);

                    boolean sectionEmpty = false;
                    switch (type) {
                        case CollectionManager.COLLECTION_TYPE_DOUBAN_MOVIE:
                            movieSize -= 1;
                            sectionEmpty = movieSize == 0;
                            break;
                        case CollectionManager.COLLECTION_TYPE_ZHIHU_STORY:
                            storySize -= 1;
                            sectionEmpty = storySize == 0;
                            break;
                        case CollectionManager.COLLECTION_TYPE_ONE_REVIEW:
                            oneReviewSize -= 1;
                            sectionEmpty = oneReviewSize == 0;
                            break;
                        case CollectionManager.COLLECTION_TYPE_ONE_ARTICLE:
                            oneArticleSize -= 1;
                            sectionEmpty = oneArticleSize == 0;
                            break;
                    }
                    if (sectionEmpty) {
                        getAdapter().remove(position - 1);
                        getAdapter().notifyItemRemoved(position - 1);
                    }
                }
            });
        }
    };
}

private void initActions() {
    onLoadLatestNews = new Action1<ZhihuDailyList>() {
        @Override
        public void call(ZhihuDailyList list) {
            initBanner(list.topStories);
            getAdapter().add(new ZhihuDateItemViewModel(R.string.special_section));
            getAdapter().add(section);
            getAdapter().add(new ZhihuDateItemViewModel(R.string.latest_daily));
            initNews(list.stories);
            hideRefreshing();
            loadSections();
            setIsLoadingMore(false);
            enableLoadMore();
        }
    };
    onRefreshLatestNews = new Action1<ZhihuDailyList>() {
        @Override
        public void call(ZhihuDailyList list) {
            getAdapter().clear();
            getAdapter().add(banner);
            getAdapter().add(new ZhihuDateItemViewModel(R.string.special_section));
            getAdapter().add(section);
            updateBanner(list.topStories);
            getAdapter().add(new ZhihuDateItemViewModel(R.string.latest_daily));
            initNews(list.stories);
            hideRefreshing();
            loadSections();
            setIsLoadingMore(false);
            enableLoadMore();
            date = new Date();          //重置日期
        }
    };
    onDailyItemClick = new Action3<String, Integer, Integer>() {
        @Override
        public void call(String id, Integer touchX, Integer touchY) {
            Intent intent = ZhihuStoryDetailActivity.intentFor(getContext());
            intent.putExtra(Constant.EXTRA_ZHIHU_STORY_ID, id);
            intent.putExtra(Constant.EXTRA_TOUCH_X, touchX);
            intent.putExtra(Constant.EXTRA_TOUCH_Y, touchY);
            getContext().startActivity(intent);
        }
    };
    onSectionItemClick = new Action2<String, String>() {
        @Override
        public void call(String id, String name) {
            Intent intent = ZhihuSectionDetailActivity.intentFor(getContext());
            intent.putExtra(Constant.EXTRA_ZHIHU_SECTION_ID, id);
            intent.putExtra(Constant.EXTRA_ZHIHU_SECTION_NAME, name);
            getContext().startActivity(intent);
        }
    };
}

public ListRecyclerAdapter<T, VH> bindViewHolder(Action3<VH, Integer, T> onBindViewHolder) {
    mOnBindViewHolder = onBindViewHolder;
    return this;
}

@SuppressWarnings("unchecked")
public static <T, R, S> Action3<T, R, S> doNothing3() {
    return (Action3<T, R, S>) HolderDoNothing3.INSTANCE;
}

public void setRecordCounterListener(Action3 action) {
    this.recordCounterListener = action;
}

@Test
public void testCheckPause() throws IOException, InterruptedException {
    TopicCommand.createTopic(zk.getZkClient(),
            new TopicCommand.TopicCommandOptions(new String[]{
                    "--zookeeper", "dummy", "--create", "--topic", TOPIC_NAME + "check_pause",
                    "--replication-factor", "2", "--partitions", "1"}));
    String description = "{\n" +
            "    \"type\": \"kafka\",\n" +
            "    \"client.id\": \"kafkasink\",\n" +
            "    \"bootstrap.servers\": \"" + kafkaServer.getBrokerListStr() + "\",\n" +
            "    \"acks\": 1,\n" +
            "    \"buffer.memory\": 1000,\n" +
            "    \"batch.size\": 1000\n" +
            "}";


    final KafkaSink sink = jsonMapper.readValue(description, new TypeReference<Sink>(){});
    sink.open();

    final AtomicBoolean exceptionCaught = new AtomicBoolean(false);
    final AtomicBoolean checkPaused = new AtomicBoolean(false);
    final AtomicBoolean pending = new AtomicBoolean(false);
    final CountDownLatch latch = new CountDownLatch(1);

    sink.setRecordCounterListener(new Action3<Long, Long, Long>() {

        @Override
        public void call(Long queued, Long sent, Long dropped) {
            if (dropped > 0) {
                exceptionCaught.set(true);
                if (sink.checkPause() > 0) {
                    checkPaused.set(true);
                }
                if (sink.getNumOfPendingMessages() > 0) {
                    pending.set(true);
                }
                latch.countDown();
            }
        }
    });
    for (int i = 0; i < 100; ++i) {
        sink.writeTo(new DefaultMessageContainer(new Message(TOPIC_NAME + "check_pause", getBigData()), jsonMapper));
    }
    assertTrue(latch.await(10, TimeUnit.SECONDS));
    assertTrue(exceptionCaught.get());
    assertTrue(checkPaused.get());
    assertTrue(pending.get());
}

@Test
public void testStartWithKafkaOutage() throws Throwable {
    String topicName = TOPIC_NAME + "kafkaoutage";

    TopicCommand.createTopic(zk.getZkClient(),
        new TopicCommand.TopicCommandOptions(new String[]{
            "--zookeeper", "dummy", "--create", "--topic", topicName,
            "--replication-factor", "2", "--partitions", "1"}));

    String[] brokerList = kafkaServer.getBrokerListStr().split(",");
    int port1 = Integer.parseInt(brokerList[0].split(":")[1]);
    int port2 = Integer.parseInt(brokerList[1].split(":")[1]);

    String description = "{\n" +
        "    \"type\": \"kafka\",\n" +
        "    \"client.id\": \"kafkasink\",\n" +
        "    \"bootstrap.servers\": \"" + kafkaServer.getBrokerListStr() + "\",\n" +
        "    \"acks\": 1\n" +
        "    }" +
        "}";

    kafkaServer.shutdown();


    final KafkaSink sink = jsonMapper.readValue(description, new TypeReference<Sink>(){});
    sink.open();

    final int msgCount = 10000;
    final CountDownLatch latch = new CountDownLatch(1);
    sink.setRecordCounterListener(new Action3<Long, Long, Long>() {

        @Override
        public void call(Long queued, Long sent, Long dropped) {
            if (sent == msgCount - sink.droppedRecords.get()) {
                latch.countDown();
            }
        }
    });

    sendMessages(topicName, sink, msgCount);

    kafkaServer.startServer(port1, port2); // running up
    assertTrue(latch.await(10, TimeUnit.SECONDS));

    sendMessages(topicName, sink, msgCount);
    sink.close();

    checkConsumer(topicName, 2 * msgCount - (int) sink.droppedRecords.get());
}

/**
 * Combines provided actions into a single action stream
 *
 * @param a1 Action
 * @return Single action stream combined from provided actions
 */
@CheckResult
@NonNull
public static <T1, T2, T3> Action3<? super T1, ? super T2, ? super T3> combine(@NonNull Action3<? super T1, ? super T2, ? super T3> a1) {
    return a1;
}