import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Accept an image upload via POST and notify a Reactor that the image needs to be thumbnailed. Asynchronously respond
 * to the client when the thumbnailing has completed.
 *
 * @param channel
 *     the channel on which to send an HTTP response
 * @param thumbnail
 *     a reference to the shared thumbnail path
 * @param reactor
 *     the Reactor on which to publish events
 *
 * @return a consumer to handle HTTP requests
 */
public static Consumer<FullHttpRequest> thumbnailImage(NetChannel<FullHttpRequest, FullHttpResponse> channel,
                                                       AtomicReference<Path> thumbnail,
                                                       Reactor reactor) {
  return req -> {
    if (req.getMethod() != HttpMethod.POST) {
      channel.send(badRequest(req.getMethod() + " not supported for this URI"));
      return;
    }

    // write to a temp file
    Path imgIn = null;
    try {
      imgIn = readUpload(req.content());
    } catch (IOException e) {
      throw new IllegalStateException(e.getMessage(), e);
    }

    // Asynchronously thumbnail the image to 250px on the long side
    reactor.sendAndReceive("thumbnail", Event.wrap(imgIn), ev -> {
      thumbnail.set(ev.getData());
      channel.send(redirect());
    });
  };
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Instantiates a new annotation audit output stream.
 *
 * @param outputStream            the output stream
 * @param transformer the transformer
 */
public AsyncAnnotationAuditOutputStream(final AuditOutputStream<AuditEvent> outputStream, final AnnotationTransformer transformer) {
    ENV = new Environment();
    this.outputStream = outputStream;
    b = new Boundary();
    
    annotationDeferred = Streams.<AnnotationAuditEvent> defer().env(ENV).dispatcher(Environment.RING_BUFFER).get();
    Stream<AnnotationAuditEvent> annostream = annotationDeferred.compose();
    annostream.consume(b.bind(new Consumer<AnnotationAuditEvent>() {
        @Override
        public void accept(AnnotationAuditEvent annotationEvent) {
            AuditEvent event = transformer.transformToEvent(annotationEvent);
            // event is null if annotation is not found
            if (event != null) {
                outputStream.write(event);
            }
        }
    }));
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Instantiates a new async audit output stream.
 *
 * @param outputStream the output stream
 * @param annotationStream the annotation stream
 */
public AsyncAuditOutputStream(final AuditOutputStream<AuditEvent> outputStream, final AuditOutputStream<AnnotationAuditEvent> annotationStream) {
    this.outputStream = outputStream;
    this.annotationStream = annotationStream;
    
    ENV = new Environment();
    b = new Boundary();
    deferred = Streams.<AuditEvent> defer().env(ENV).dispatcher(Environment.RING_BUFFER).get();
    Stream<AuditEvent> stream = deferred.compose();
    stream.consume(b.bind(new Consumer<AuditEvent>() {
        @Override
        public void accept(AuditEvent event) {
            outputStream.write(event);
        }
    }));
    
    batchBoundry = new Boundary();
    batchDeferred = Streams.<EventBatch> defer().env(ENV).dispatcher(Environment.RING_BUFFER).get();
    Stream<EventBatch> batchStream = batchDeferred.compose();
    batchStream.consume(batchBoundry.bind(new Consumer<EventBatch>() {
        @Override
        public void accept(EventBatch batch) {
            outputStream.writeBatch(batch);
        }
    }));
}
 

import reactor.function.Consumer; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
    Environment env = new Environment();
    final TradeServer server = new TradeServer();

    // Use a Reactor to dispatch events using the default Dispatcher
    Reactor reactor = Reactors.reactor()
            .env(env)
            .dispatcher("ringBuffer")
            .get();

    String topic = "trade.execute";

    // For each Trade event, execute that on the server
    reactor.on($(topic), new Consumer<Event<Trade>>() {
        @Override
        public void accept(Event<Trade> tradeEvent) {
            server.execute(tradeEvent.getData());

            // Since we're async, for this test, use a latch to tell when we're done
            latch.countDown();
        }
    });

    // Start a throughput timer
    startTimer();

    // Publish one event per trade
    for (int i = 0; i < totalTrades; i++) {
        // Pull next randomly-generated Trade from server
        Trade t = server.nextTrade();

        // Notify the Reactor the event is ready to be handled
        reactor.notify(topic, Event.wrap(t));
    }

    // Stop throughput timer and output metrics
    endTimer();

    server.stop();
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Creates a new SubscriptionManager
 * @param config The extracted configuration
 */
public SubscriptionManager(final Properties config) {
	reactor = SingletonEnvironment.getInstance().getDefaultReactor();
	reactor.on(Selectors.object("subscription-terminated"), new Consumer<Event<Subscription>>(){
		@Override
		public void accept(Event<Subscription> t) {
			final Subscription termSub = t.getData();
			log.info("Terminated Subscription: [{}]", termSub);				
			allSubscriptions.remove(termSub.pattern);				
		}
	});
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Respond to GET requests and serve the thumbnailed image, a reference to which is kept in the given {@literal
 * AtomicReference}.
 *
 * @param channel
 *     the channel on which to send an HTTP response
 * @param thumbnail
 *     a reference to the shared thumbnail path
 *
 * @return a consumer to handle HTTP requests
 */
public static Consumer<FullHttpRequest> serveThumbnailImage(NetChannel<FullHttpRequest, FullHttpResponse> channel,
                                                            AtomicReference<Path> thumbnail) {
  return req -> {
    if (req.getMethod() != HttpMethod.GET) {
      channel.send(badRequest(req.getMethod() + " not supported for this URI"));
    } else {
      try {
        channel.send(serveImage(thumbnail.get()));
      } catch (IOException e) {
        throw new IllegalStateException(e.getMessage(), e);
      }
    }
  };
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Respond to errors occurring on a Reactor by redirecting them to the client via an HTTP 500 error response.
 *
 * @param channel
 *     the channel on which to send an HTTP response
 *
 * @return a consumer to handle HTTP requests
 */
public static Consumer<Throwable> errorHandler(NetChannel<FullHttpRequest, FullHttpResponse> channel) {
  return ev -> {
    DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                                                               HttpResponseStatus.INTERNAL_SERVER_ERROR);
    resp.content().writeBytes(ev.getMessage().getBytes());
    resp.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain");
    resp.headers().set(HttpHeaders.Names.CONTENT_LENGTH, resp.content().readableBytes());
    channel.send(resp);
  };
}
 

import reactor.function.Consumer; //导入依赖的package包/类
/**
 * Respond to GET requests and serve the thumbnailed image, a reference to which is kept in the given {@literal
 * AtomicReference}.
 *
 * @param channel
 *     the channel on which to send an HTTP response
 *
 * @return a consumer to handle HTTP requests
 */
public static Consumer<FullHttpRequest> serveThumbnailImage(NetChannel<FullHttpRequest, FullHttpResponse> channel) {
  return req -> {
    channel.send(request("Hello World!"));
  };
}