import reactor.fn.Consumer; //导入依赖的package包/类
@Override
public Publisher<Void> apply(ChannelStream<Message<P>, Message<P>> channelStream) {
	Promise<Void> closePromise = Promises.prepare();
	this.connectionHandler.afterConnected(new Reactor2TcpConnection<P>(channelStream, closePromise));
	channelStream
			.finallyDo(new Consumer<Signal<Message<P>>>() {
				@Override
				public void accept(Signal<Message<P>> signal) {
					if (signal.isOnError()) {
						connectionHandler.handleFailure(signal.getThrowable());
					}
					else if (signal.isOnComplete()) {
						connectionHandler.afterConnectionClosed();
					}
				}
			})
			.consume(new Consumer<Message<P>>() {
				@Override
				public void accept(Message<P> message) {
					connectionHandler.handleMessage(message);
				}
			});

	return closePromise;
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public static void main(String[] args) {
	//Initialize context and get default dispatcher
	Environment.initializeIfEmpty();

	//RingBufferDispatcher with 8192 slots by default
	Dispatcher dispatcher = Environment.sharedDispatcher();

	//Create a callback
	Consumer<Integer> c = data ->
	        System.out.format("some data arrived: %s\n", data);

	//Create an error callback
	Consumer<Throwable> errorHandler = ex -> ex.printStackTrace();

	//Dispatch data asynchronously
	dispatcher.dispatch(1234, c, errorHandler);
	dispatcher.dispatch(5678, c, errorHandler);

	Environment.terminate();
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();
	CountDownLatch latch = new CountDownLatch(5);

	Dispatcher trreadPoolDispatcher = new ThreadPoolExecutorDispatcher(5, 128);

	Consumer<String> consumer = ev -> {
	    LOG.info("Hello " + ev + " from thread: " +Thread.currentThread() + "\n");
	    latch.countDown(); 
	};

	Consumer<Throwable> errorConsumer = error ->
	    error.printStackTrace();

	// a task is submitted to the thread pool dispatcher
	Stream<String> stream = Streams.just("One", "Two", "Three", "Four", "Five", "Six", "Seven");
	stream.dispatchOn(env).consume(ev -> {
				System.out.println(ev);
				trreadPoolDispatcher.dispatch(ev, consumer, errorConsumer);
			});

	latch.await(15, TimeUnit.SECONDS); // Wait for task to execute

}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();
	CountDownLatch latch = new CountDownLatch(5);

	DispatcherSupplier supplier = Environment.newCachedDispatchers(3, "myPool");
	
	Consumer<String> consumer = ev -> {
	    LOG.info("Hello " + ev + " from thread: " +Thread.currentThread() + "\n");
	    latch.countDown(); 
	};

	Consumer<Throwable> errorConsumer = error ->
	    error.printStackTrace();

	// a task is submitted to the thread pool dispatcher
	Stream<String> stream = Streams.just("One", "Two", "Three", "Four", "Five", "Six", "Seven");
	stream.dispatchOn(env).partition().consume(
		groupStream -> 
			groupStream.dispatchOn(supplier.get()).consume(consumer, errorConsumer)
	);
	latch.await(15, TimeUnit.SECONDS); // Wait for task to execute

}
 

import reactor.fn.Consumer; //导入依赖的package包/类
@Override
protected void init(VaadinRequest vaadinRequest) {

  final VerticalLayout layout = new VerticalLayout();
  layout.setMargin(true);
  layout.setSpacing(true);
  layout.addComponent(labelTime);
  setContent(layout);

  eventBus.on(Selectors.$(TOPIC), new Consumer<reactor.bus.Event<LocalTime>>() {
    @Override
    public void accept(reactor.bus.Event<LocalTime> event) {
      access(() -> labelTime.setValue(event.getData().format(dateTimeFormatter)));
    }
  });
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public final <E> void tryDispatch(E event, Consumer<E> eventConsumer, Consumer<Throwable> errorConsumer) {
	Assert.isTrue(alive(), "This Dispatcher has been shut down.");
	boolean isInContext = inContext();
	Task task;
	if (isInContext) {
		task = allocateRecursiveTask();
	} else {
		try {
			task = tryAllocateTask();
		} catch (InsufficientCapacityException e) {
			throw new IllegalStateException("Error in task allocation");
		}
	}

	task.setData(event).setErrorConsumer(errorConsumer).setEventConsumer(eventConsumer);

	if (!isInContext) {
		execute(task);
	}
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public final <E> void dispatch(E event, Consumer<E> eventConsumer, Consumer<Throwable> errorConsumer) {

		Assert.isTrue(alive(), "This Dispatcher has been shut down.");
		Assert.isTrue(eventConsumer != null, "The signal consumer has not been passed.");
		boolean isInContext = inContext();
		Task task;
		if (isInContext) {
			task = allocateRecursiveTask();
		} else {
			task = allocateTask();
		}

		task.setData(event).setErrorConsumer(errorConsumer).setEventConsumer(eventConsumer);

		if (!isInContext) {
			execute(task);
		}
	}
 

import reactor.fn.Consumer; //导入依赖的package包/类
@PostConstruct
private void createProcessor() {
    processor = RingBufferProcessor.create();
    Stream stream = Streams.wrap(processor);
    stream.buffer(1, TimeUnit.SECONDS).consume(new Consumer<List<Long>>() {
        @Override
        public void accept(List<Long> repositoryIds) {
            for (Long repositoryId : Sets.newHashSet(repositoryIds)) {
                setRepositoryStatsOutOfDate(repositoryId);
            }
        }
    });
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public void execute(final Runnable command) {
	dispatch(null, new Consumer<Object>() {
		@Override
		public void accept(Object ev) {
			command.run();
		}
	}, null);
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
@Test
public void testInitialize() {
    List<FlowConfiguration<?>> flowConfigs = new ArrayList<>();
    flowConfigs.add(new StackSyncFlowConfig());
    flowConfigs.add(new StackTerminationFlowConfig());
    given(this.flowConfigs.stream()).willReturn(flowConfigs.stream());
    underTest.init();
    verify(reactor, times(1)).on(any(Selector.class), any(Consumer.class));
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
@Test
    public void tags() throws IOException {

        final Broadcaster<Object> broadcaster = SerializedBroadcaster.create();

        Processor processor = new TopTags(1,10);
        Stream<?> outputStream = processor.process(broadcaster);


        outputStream.consume(new Consumer<Object>() {
            @Override
            public void accept(Object o) {
                System.out.println("processed : " + o);
            }
            //TODO - expect
//            processed : {"id":"55786760-7472-065d-8e62-eb83260948a4","timestamp":1422399628134,"hashtag":"AndroidGames","count":1}
//            processed : {"id":"bd99050f-abfa-a239-c09a-f2fe721daafb","timestamp":1422399628182,"hashtag":"Android","count":1}
//            processed : {"id":"10ce993c-fd57-322d-efa1-16f810918187","timestamp":1422399628184,"hashtag":"GameInsight","count":1}
        });

        ClassPathResource resource = new ClassPathResource("tweets.json");
        Scanner scanner = new Scanner(resource.getInputStream());
        while (scanner.hasNext()) {
            String tweet = scanner.nextLine();
            broadcaster.onNext(tweet);
            //simulateLatency();
        }
        //System.in.read();

    }
 

import reactor.fn.Consumer; //导入依赖的package包/类
@Override
public Function<Buffer, Message<byte[]>> decoder(final Consumer<Message<byte[]>> messageConsumer) {
	return new DecodingFunction(this.stompDecoder, messageConsumer);
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public DecodingFunction(StompDecoder decoder, Consumer<Message<byte[]>> next) {
	this.decoder = decoder;
	this.messageConsumer = next;
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public <E extends ApplicationEvent> void register(final Selector channel, final Consumer<Event<E>> eventListener) {
    notNull(eventListener);
    notNull(channel);
    eventBus.on(channel, eventListener);
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
@Override
public Function<Buffer, Buffer> decoder(Consumer<Buffer> next) {
	return null;
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public Task setEventConsumer(Consumer<?> eventConsumer) {
	this.eventConsumer = eventConsumer;
	return this;
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
public Task setErrorConsumer(Consumer<Throwable> errorConsumer) {
	this.errorConsumer = errorConsumer;
	return this;
}
 

import reactor.fn.Consumer; //导入依赖的package包/类
/**
 * Creates a new {@literal RingBufferDispatcher} with the given {@code name}
 * . It will use a {@link RingBuffer} with {@code bufferSize} slots,
 * configured with a producer type of {@link ProducerType#MULTI MULTI} and a
 * {@link BlockingWaitStrategy blocking wait. A given @param
 * uncaughtExceptionHandler} will catch anything not handled e.g. by the
 * owning {@code reactor.bus.EventBus} or {@code reactor.rx.Stream}.
 *
 * @param name
 *            The name of the dispatcher
 * @param bufferSize
 *            The size to configure the ring buffer with
 * @param uncaughtExceptionHandler
 *            The last resort exception handler
 */
public RingBufferLocalDispatcher(String name, int bufferSize, final Consumer<Throwable> uncaughtExceptionHandler) {
	this(name, bufferSize, uncaughtExceptionHandler, ProducerType.MULTI, new BlockingWaitStrategy());

}