import reactor.core.Cancellation; //导入依赖的package包/类
/**
 * A tailable cursor streams data using {@link Flux} as it arrives inside the capped collection.
 */
@Test
public void shouldStreamDataWithTailableCursor() throws Exception {

	Cancellation cancellation = repository.findWithTailableCursorBy() //
			.doOnNext(System.out::println) //
			.doOnComplete(() -> System.out.println("Complete")) //
			.doOnTerminate(() -> System.out.println("Terminated")) //
			.subscribe();

	Thread.sleep(100);

	repository.save(new Person("Tuco", "Salamanca", 33)).subscribe();
	Thread.sleep(100);

	repository.save(new Person("Mike", "Ehrmantraut", 62)).subscribe();
	Thread.sleep(100);

	cancellation.dispose();

	repository.save(new Person("Gus", "Fring", 53)).subscribe();
	Thread.sleep(100);
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
private CFStreamingLogToken internalStreamLogs(String appName, CFApplicationLogListener listener,
		boolean recentLogs) throws CoreException {
	CloudFoundryOperations operations = getV2Operations();
	V2LogListener v2Listener = asV2LogListener(listener);
	Flux<LogMessage> stream = operations.applications()
			.logs(LogsRequest.builder().name(appName).recent(recentLogs).build());
	final Cancellation cancellation = stream.subscribe(v2Listener::onMessage, v2Listener::onError);
	return new CFStreamingLogToken() {

		@Override
		public void cancel() {
			cancellation.dispose();
		}
	};
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
@Override
public ScheduledFuture<?> scheduleAtFixedRate(final Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
	long initialDelayInMs = convertToMillis(initialDelay, unit);
	long periodInMs = convertToMillis(period, unit);
	final AtomicReference<Cancellation> registration = new AtomicReference<>();

	final Runnable task = () -> {
		try {
			command.run();
		} catch (Throwable t) {
			log.error(t.getMessage(), t);
			Cancellation reg;
			if (null != (reg = registration.get())) {
				reg.dispose();
			}
		}
	};

	final Runnable consumer = () -> execute(task);

	final ScheduledFutureTask<?> future = new ScheduledFutureTask<Object>(task, null, initialDelay);
	registration.set(timer.schedulePeriodically(consumer, initialDelayInMs, periodInMs, TimeUnit.MILLISECONDS));
	return future;
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(final Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
	final long initialDelayInMs = convertToMillis(initialDelay, unit);
	final long delayInMs = convertToMillis(initialDelay, unit);
	final ScheduledFutureTask<?> future = new ScheduledFutureTask<Object>(command, null, initialDelayInMs);

	final AtomicReference<Cancellation> registration = new AtomicReference<>();

	final Runnable consumer = new Runnable() {
		final Runnable self = this;

		@Override
		public void run() {
			execute(() -> {
				try {
					future.run();
					timer.schedule(self, delayInMs, TimeUnit.MILLISECONDS);
				} catch (Throwable t) {
					log.error(t.getMessage(), t);
					Cancellation reg;
					if (null != (reg = registration.get())) {
						reg.dispose();
					}
				}
			});
		}
	};

	registration.set(timer.schedule(consumer, initialDelayInMs, TimeUnit.MILLISECONDS));
	return future;
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
@Override
public boolean subscribe(final MessageHandler handler) {
	Consumer<Message<?>> consumer = handler::handleMessage;
	Cancellation c = Flux.from(processor).subscribe(consumer);
	messageHandlerConsumers.put(handler, c);

	return true;
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public boolean unsubscribe(MessageHandler handler) {
	Cancellation control = messageHandlerConsumers.remove(handler);
	if (null == control) {
		return false;
	}
	control.dispose();
	return true;
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
/**
 * @param consumer
 * @return
 * @see reactor.core.publisher.Mono#subscribe(java.util.function.Consumer)
 */
public final Cancellation subscribe(Consumer<? super T> consumer) {
    return boxed.subscribe(consumer);
}
 

import reactor.core.Cancellation; //导入依赖的package包/类
/**
 * @return
 * @see reactor.core.publisher.Flux#subscribe()
 */
public final Cancellation subscribe() {
    return boxed.subscribe();
}