import reactor.fn.Function; //导入依赖的package包/类
@Override
public boolean equals(Object o) {
    if (o == this)
        return true;
    if (!(o instanceof RedissonListReactive))
        return false;

    Stream<Object> e1 = Streams.wrap((Publisher<Object>)iterator());
    Stream<Object> e2 = Streams.wrap(((RedissonListReactive<Object>) o).iterator());
    Long count = Streams.merge(e1, e2).groupBy(new Function<Object, Object>() {
        @Override
        public Object apply(Object t) {
            return t;
        }
    }).count().next().poll();

    boolean res = count.intValue() == Streams.wrap(size()).next().poll();
    res &= count.intValue() == Streams.wrap(((RedissonListReactive<Object>) o).size()).next().poll();
    return res;
}
 

import reactor.fn.Function; //导入依赖的package包/类
private Function<Entry<K, V>, Boolean> mapFunction(final Map<?, ?> m) {
    return new Function<Map.Entry<K, V>, Boolean>() {
        @Override
        public Boolean apply(Entry<K, V> e) {
            K key = e.getKey();
            V value = e.getValue();
            if (value == null) {
                if (!(m.get(key)==null && m.containsKey(key)))
                    return false;
            } else {
                if (!value.equals(m.get(key)))
                    return false;
            }
            return true;
        }
    };
}
 

import reactor.fn.Function; //导入依赖的package包/类
@Test
public void testBindZero() throws Exception {
	Environment.initializeIfEmpty().assignErrorJournal();

	HttpServer<Buffer, Buffer> httpServer = NetStreams
			.httpServer(new Function<HttpServerSpec<Buffer, Buffer>, HttpServerSpec<Buffer, Buffer>>() {

				@Override
				public HttpServerSpec<Buffer, Buffer> apply(HttpServerSpec<Buffer, Buffer> server) {
					return server
							.codec(new GpfdistCodec())
							.listen(0);
				}
			});
	httpServer.start().awaitSuccess();
	InetSocketAddress address = httpServer.getListenAddress();
	assertThat(address, notNullValue());
	assertThat(address.getPort(), not(0));
	httpServer.shutdown();
}
 

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

import reactor.fn.Function; //导入依赖的package包/类
@Override
public Function<Message<byte[]>, Buffer> encoder() {
	return this.encodingFunction;
}
 

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

import reactor.fn.Function; //导入依赖的package包/类
private HttpServer<Buffer, Buffer> createProtocolListener()
		throws Exception {

	final Stream<Buffer> stream = Streams
	.wrap(processor)
	.window(flushCount, flushTime, TimeUnit.SECONDS)
	.flatMap(new Function<Stream<Buffer>, Publisher<Buffer>>() {

		@Override
		public Publisher<Buffer> apply(Stream<Buffer> t) {

			return t.reduce(new Buffer(), new BiFunction<Buffer, Buffer, Buffer>() {

				@Override
				public Buffer apply(Buffer prev, Buffer next) {
					return prev.append(next);
				}
			});
		}
	})
	.process(RingBufferWorkProcessor.<Buffer>create("gpfdist-sink-worker", 8192, false));

	HttpServer<Buffer, Buffer> httpServer = NetStreams
			.httpServer(new Function<HttpServerSpec<Buffer, Buffer>, HttpServerSpec<Buffer, Buffer>>() {

				@Override
				public HttpServerSpec<Buffer, Buffer> apply(HttpServerSpec<Buffer, Buffer> server) {
					return server
							.codec(new GpfdistCodec())
							.listen(port);
				}
			});

	httpServer.get("/data", new ReactorChannelHandler<Buffer, Buffer, HttpChannel<Buffer,Buffer>>() {

		@Override
		public Publisher<Void> apply(HttpChannel<Buffer, Buffer> request) {
			request.responseHeaders().removeTransferEncodingChunked();
			request.addResponseHeader("Content-type", "text/plain");
			request.addResponseHeader("Expires", "0");
			request.addResponseHeader("X-GPFDIST-VERSION", "Spring Dataflow");
			request.addResponseHeader("X-GP-PROTO", "1");
			request.addResponseHeader("Cache-Control", "no-cache");
			request.addResponseHeader("Connection", "close");

			return request.writeWith(stream
					.take(batchCount)
					.timeout(batchTimeout, TimeUnit.SECONDS, Streams.<Buffer>empty())
					.concatWith(Streams.just(Buffer.wrap(new byte[0]))))
					.capacity(1l);
		}
	});

	httpServer.start().awaitSuccess();
	log.info("Server running using address=[" + httpServer.getListenAddress() + "]");
	localPort = httpServer.getListenAddress().getPort();
	return httpServer;
}