import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Test
public void test4() {
    SomeFeed<PriceTick> feed = new SomeFeed<>();
    Flux<Object> priceFlux = Flux.create(emitter ->
    {
        SomeListener l = new SomeListener() {
            @Override
            public void priceTick(PriceTick event) {
                emitter.next(event);
            }

            @Override
            public void error(Throwable throwable) {
                emitter.error(throwable);
            }

        };
        feed.register(l);
    }, FluxSink.OverflowStrategy.LATEST);
    ConnectableFlux<Object> connectableFlux = priceFlux.publish();
    connectableFlux.connect();
    connectableFlux.subscribe(System.out::println);
}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Test
public void advancedConnectable() throws InterruptedException {
	Flux<Integer> source = Flux.range(1, 3)
	                           .doOnSubscribe(s -> System.out.println("subscribed to source"));

	ConnectableFlux<Integer> co = source.publish();

	co.subscribe(System.out::println, e -> {}, () -> {});
	co.subscribe(System.out::println, e -> {}, () -> {});

	System.out.println("done subscribing");
	Thread.sleep(500);
	System.out.println("will now connect");

	co.connect();
}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Test
public void test() throws InterruptedException {
    SomeFeed<PriceTick> feed = new SomeFeed<>();
    Flux<PriceTick> flux =
            Flux.create(emitter ->
            {
                SomeListener listener = new SomeListener() {
                    @Override
                    public void priceTick(PriceTick event) {
                        emitter.next(event);
                        if (event.isLast()) {
                            emitter.complete();
                        }
                    }

                    @Override
                    public void error(Throwable e) {
                        emitter.error(e);
                    }
                };
                        feed.register(listener);
    }, FluxSink.OverflowStrategy.LATEST);
    ConnectableFlux connectable = flux.publish();
    connectable.subscribe(x -> System.out.println("1st " + x));
    Thread.sleep(1000);
    connectable.subscribe(x -> System.out.println("2nd " + x));
    connectable.connect();

}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Test
public void testMultiReceiver() throws Exception {
	Hooks.onOperatorDebug();
	try {

		ConnectableFlux<?> t = Flux.empty()
		    .then(Mono.defer(() -> {
			    throw new RuntimeException();
		    })).flux().publish();

		t.map(d -> d).subscribe(null,
				e -> Assert.assertTrue(e.getSuppressed()[0].getMessage().contains
						("\t|_\tFlux.publish")));

		t.filter(d -> true).subscribe(null, e -> Assert.assertTrue(e.getSuppressed()[0].getMessage().contains
				("\t\t|_\tFlux.publish")));
		t.distinct().subscribe(null, e -> Assert.assertTrue(e.getSuppressed()[0].getMessage().contains
				("\t\t\t|_\tFlux.publish")));

		t.connect();
	}
	finally {
		Hooks.resetOnOperatorDebug();
	}
}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Override
public ConnectableFlux<String> freeFlowEmps() {
	 Vector<String> rosterNames = new Vector<>();
	 Function<Employee, String> familyNames = (emp) -> emp.getLastName().toUpperCase();
	 ConnectableFlux<String> flowyNames = Flux.fromIterable(employeeDaoImpl.getEmployees()).log().map(familyNames).cache().publish();
	// flowyNames.subscribe(System.out::println);
	 flowyNames.subscribe(rosterNames::add);
	 System.out.println(rosterNames);
	return flowyNames;
}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
@Test
public void create() throws InterruptedException {
	SomeFeed<PriceTick> feed = new SomeFeed<>();
	Flux<PriceTick> flux =
			Flux.create(emitter ->
			{
				SomeListener listener = new SomeListener() {
					@Override
					public void priceTick(PriceTick event) {
						emitter.next(event);
						if (event.isLast()) {
							emitter.complete();
						}
					}

					@Override
					public void error(Throwable e) {
						emitter.error(e);
					}};
				feed.register(listener);
			}, FluxSink.OverflowStrategy.BUFFER);

	ConnectableFlux<PriceTick> hot = flux.publish();

	hot.subscribe(priceTick -> System.out.printf("%s %4s %6.2f%n", priceTick
			.getDate(), priceTick.getInstrument(), priceTick.getPrice()));

	hot.subscribe(priceTick -> System.out.println(priceTick.getSequence() +
			": " + priceTick.getInstrument()));
	hot.connect();
	Thread.sleep(5000);
}
 

import reactor.core.publisher.ConnectableFlux; //导入依赖的package包/类
public ConnectableFlux<String> freeFlowEmps();