Java ReplayProcessor类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中reactor.core.publisher.ReplayProcessor类的典型用法代码示例。如果您正苦于以下问题：Java ReplayProcessor类的具体用法？Java ReplayProcessor怎么用？Java ReplayProcessor使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

ReplayProcessor类属于reactor.core.publisher包，在下文中一共展示了ReplayProcessor类的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: echo

import reactor.core.publisher.ReplayProcessor; //导入依赖的package包/类
@Test
public void echo() throws Exception {
	int count = 100;
	Flux<String> input = Flux.range(1, count).map(index -> "msg-" + index);
	ReplayProcessor<Object> output = ReplayProcessor.create(count);

	client.execute(getUrl("/echo"),
			session -> {
				logger.debug("Starting to send messages");
				return session
						.send(input.doOnNext(s -> logger.debug("outbound " + s)).map(session::textMessage))
						.thenMany(session.receive().take(count).map(WebSocketMessage::getPayloadAsText))
						.subscribeWith(output)
						.doOnNext(s -> logger.debug("inbound " + s))
						.then()
						.doOnSuccessOrError((aVoid, ex) ->
								logger.debug("Done with " + (ex != null ? ex.getMessage() : "success")));
			})
			.block(Duration.ofMillis(5000));

	assertEquals(input.collectList().block(Duration.ofMillis(5000)),
			output.collectList().block(Duration.ofMillis(5000)));
}

开发者ID:spring-cloud，项目名称:spring-cloud-gateway，代码行数:24，代码来源:WebSocketIntegrationTests.java

示例2: main

import reactor.core.publisher.ReplayProcessor; //导入依赖的package包/类
public static final void main(String[] args) throws URISyntaxException {

        WebSocketClient client = new ReactorNettyWebSocketClient();
//        client.execute(new URI("ws://localhost:8080/echo"), (WebSocketSession session) -> {
//            session.send().log().;
//        });
        
        int count = 100;
		Flux<String> input = Flux.range(1, count).map(index -> "msg-" + index);
		ReplayProcessor<Object> output = ReplayProcessor.create(count);

		client.execute(new URI("ws://localhost:8080/echo"),
				session -> {
					log.debug("Starting to send messages");
					return session
							.send(input.doOnNext(s -> log.debug("outbound " + s)).map(session::textMessage))
							.thenMany(session.receive().take(count).map(WebSocketMessage::getPayloadAsText))
							.subscribeWith(output)
							.doOnNext(s -> log.debug("inbound " + s))
							.then()
							.doOnTerminate((aVoid, ex) ->
									log.debug("Done with " + (ex != null ? ex.getMessage() : "success")));
				})
				.block(Duration.ofMillis(5000));

//		assertEquals(input.collectList().block(Duration.ofMillis(5000)),
//				output.collectList().block(Duration.ofMillis(5000)));
//        client.execute(new URI("ws://localhost:8080/echo")), session -> {
//            session.
//        }
//        ).blockMillis(5000);
    }

开发者ID:hantsy，项目名称:spring-reactive-sample，代码行数:33，代码来源:WebSocketDemoClient.java

示例3: analyticsTest

import reactor.core.publisher.ReplayProcessor; //导入依赖的package包/类
@Test
public void analyticsTest() throws Exception {
	ReplayProcessor<Integer> source = ReplayProcessor.create();

	long avgTime = 50l;

	Mono<Long> result = source
			.log("delay")
			.publishOn(asyncGroup)
	                          .delayElements(Duration.ofMillis(avgTime))

	                          .elapsed()
	                          .skip(1)
	                          .groupBy(w -> w.getT1())
							  .flatMap(w -> w.count().map(c -> Tuples.of(w.key(), c)))
	                          .log("elapsed")
	                          .collectSortedList(Comparator.comparing(Tuple2::getT1))
	                          .flatMapMany(Flux::fromIterable)
	                          .reduce(-1L, (acc, next) -> acc > 0l ? ((next.getT1() + acc) / 2) : next.getT1())
	                          .log("reduced-elapsed")
	                          .cache();

	source.subscribe();

	for (int j = 0; j < 10; j++) {
		source.onNext(1);
	}
	source.onComplete();

	Assert.assertTrue(result.block(Duration.ofSeconds(5)) >= avgTime * 0.6);
}

开发者ID:reactor，项目名称:reactor-core，代码行数:32，代码来源:FluxTests.java

示例4: combineWithOneElement

import reactor.core.publisher.ReplayProcessor; //导入依赖的package包/类
@Test
public void combineWithOneElement() throws InterruptedException, TimeoutException {
	AtomicReference<Object> ref = new AtomicReference<>(null);

	Phaser phaser = new Phaser(2);

	Flux<Object> s1 = ReplayProcessor.cacheLastOrDefault(new Object())
	                                 .publishOn(asyncGroup);
	Flux<Object> s2 = ReplayProcessor.cacheLastOrDefault(new Object())
	                                .publishOn(asyncGroup);

	// The following works:
	//List<Flux<Object>> list = Arrays.collectList(s1);
	// The following fails:
	List<Flux<Object>> list = Arrays.asList(s1, s2);

	Flux.combineLatest(list, t -> t)
	       .log()
	       .doOnNext(obj -> {
		       ref.set(obj);
		       phaser.arrive();
	       })
	       .subscribe();

	phaser.awaitAdvanceInterruptibly(phaser.arrive(), 1, TimeUnit.SECONDS);
	Assert.assertNotNull(ref.get());
}

开发者ID:reactor，项目名称:reactor-core，代码行数:28，代码来源:FluxTests.java

示例5: whenProcessorIsStreamed

import reactor.core.publisher.ReplayProcessor; //导入依赖的package包/类
@Test
	public void whenProcessorIsStreamed() {
//		"When a processor is streamed"
//		given: "a source composable and a async downstream"
		ReplayProcessor<Integer> source = ReplayProcessor.create();
		Scheduler scheduler = Schedulers.newParallel("test", 2);

		try {
			Mono<List<Integer>> res = source.subscribeOn(scheduler)
			                                .delaySubscription(Duration.ofMillis(1L))
			                                .log("streamed")
			                                .map(it -> it * 2)
			                                .buffer()
			                                .publishNext();

			res.subscribe();

//		when: "the source accepts a value"
			source.onNext(1);
			source.onNext(2);
			source.onNext(3);
			source.onNext(4);
			source.onComplete();

//		then: "the res is passed on"
			assertThat(res.block()).containsExactly(2, 4, 6, 8);
		}
		finally {
			scheduler.dispose();
		}
	}

开发者ID:reactor，项目名称:reactor-core，代码行数:32，代码来源:FluxSpecTests.java