Java 类io.vertx.core.eventbus.MessageConsumer 实例源码

项目：incubator-tamaya-sandbox 文件：TamayaConfigurationProducer.java

/**
 * Registers a handler for accessing single configuration keys (input: String, reply type: String). If no
 * config value is present the consumer will reply with a NOT_FOUND failure.
 * @param address the event bus address to register.
 * @param eventBus the event bus.
 * @return the consumer registered.
 */
public static MessageConsumer<String> registerSingleConfigEntryProvider(String address, EventBus eventBus){
    MessageConsumer<String> consumer = eventBus.consumer(address);
    consumer.handler(h -> {
        String key = (String) h.body();
        if (key == null) {
            h.fail(HttpResponseStatus.BAD_REQUEST.code(), "Missing config key.");
        } else {
            String value = ConfigurationProvider.getConfiguration().getOrDefault(key, null);
            if (value != null) {
                h.reply(value);
            } else {
                h.fail(HttpResponseStatus.NOT_FOUND.code(), "Config key not found: " + key);
            }
        }
    });
    return consumer;
}

项目：incubator-tamaya-sandbox 文件：TamayaConfigurationProducer.java

/**
 * Registers a handler for accessing multiple configuration keys (input: String[] (Json),
 * reply type: Map<String,String></String,String> (Json).
 * @param address the event bus address to register.
 * @param eventBus the event bus.
 * @return the consumer registered.
 */
public static MessageConsumer<String> registerMultiConfigEntryProvider(String address, EventBus eventBus){
    MessageConsumer<String> consumer = eventBus.consumer(address);
    consumer.handler(h -> {
        String val = h.body();
        Configuration config = ConfigurationProvider.getConfiguration();
        Map<String,String> entries = new TreeMap<>();
        if(val!=null){
            String[] sections = Json.decodeValue(val, String[].class);
            for (String section : sections) {
                if(section!=null) {
                    entries.putAll(config.with(ConfigurationFunctions.section(section)).getProperties());
                }
            }
        }else{
            entries.putAll(config.getProperties());
        }
        h.reply(Json.encode(entries));
    });
    return consumer;
}

项目：Karaf-Vertx 文件：JdbcServiceVertcl.java

@Override
public void start() throws Exception {
    super.start();
    client = JDBCClient.create(vertx, dataSource);


    eventBus.registerDefaultCodec(Book.class, new BookEncoder());
    eventBus.registerDefaultCodec(Recipe.class, new RecipeEncoder());
    eventBus.registerDefaultCodec((Class<ArrayList<Book>>) (Class<?>) ArrayList.class, new ListOfBookEncoder());

    initDb();

    MessageConsumer<Object> read = eventBus.consumer("de.nierbeck.vertx.jdbc.read"); //TODO: move those addresses to a "global" dict.
    MessageConsumer<Object> write = eventBus.consumer("de.nierbeck.vertx.jdbc.write.add");
    MessageConsumer<Object> update = eventBus.consumer("de.nierbeck.vertx.jdbc.write.update");
    MessageConsumer<Object> delete = eventBus.consumer("de.nierbeck.vertx.jdbc.delete");
    read.handler(this::handleRead);
    write.handler(this::handleWrite);
    update.handler(this::handleUpdate);
    delete.handler(this::handleDelete);
}

项目：vertx-service-discovery 文件：MessageSourceExamples.java

public void example2(ServiceDiscovery discovery) {
  // Get the record
  discovery.getRecord(new JsonObject().put("name", "some-message-source-service"), ar -> {
    if (ar.succeeded() && ar.result() != null) {
      // Retrieve the service reference
      ServiceReference reference = discovery.getReference(ar.result());
      // Retrieve the service object
      MessageConsumer<JsonObject> consumer = reference.getAs(MessageConsumer.class);

      // Attach a message handler on it
      consumer.handler(message -> {
        // message handler
        JsonObject payload = message.body();
      });
    }
  });
}

项目：vertx-blueprint-microservice 文件：RawOrderDispatcher.java

@Override
public void start(Future<Void> future) throws Exception {
  super.start();
  MessageSource.<JsonObject>getConsumer(discovery,
    new JsonObject().put("name", "shopping-order-message-source"),
    ar -> {
      if (ar.succeeded()) {
        MessageConsumer<JsonObject> orderConsumer = ar.result();
        orderConsumer.handler(message -> {
          Order wrappedOrder = wrapRawOrder(message.body());
          dispatchOrder(wrappedOrder, message);
        });
        future.complete();
      } else {
        future.fail(ar.cause());
      }
    });
}

项目：vertx-graphql-service-discovery 文件：SchemaRegistrar.java

/**
 * Registers a schema definition created by the
 * {@link GraphQLService}.
 * <p>
 * The provided registration is cloned, completed with publisher-related information, registered and then returned.
 *
 * @param partialRegistration the partially completed schema registration
 * @param options             the service discovery options to add
 * @param publishedHandler    the event handler to invoke on schema published events
 * @param unpublishedHandler  the event handler to invoke on schema unpublished events
 * @return the completed schema registration
 */
protected SchemaRegistration register(
        SchemaRegistration partialRegistration, ServiceDiscoveryOptions options,
        SchemaPublishedHandler<SchemaRegistration> publishedHandler,
        SchemaUnpublishedHandler<SchemaRegistration> unpublishedHandler) {

    // First start listening to schema events.
    registerSchemaEventConsumers(options, publishedHandler, unpublishedHandler);

    // Then register service consumer created from schema definition, if it was not registered yet.
    MessageConsumer<JsonObject> serviceConsumer = registerSchemaServiceConsumer(
            partialRegistration.getRecord(), partialRegistration.getSchemaDefinition());

    // Complete the schema registration
    SchemaRegistration fullRegistration = SchemaRegistration.create(partialRegistration.getDiscovery(), options,
            partialRegistration.getRecord(), partialRegistration.getSchemaDefinition(), serviceConsumer);

    return super.register(options.getName(), fullRegistration);
}

项目：vertx-stomp 文件：EventBusBridge.java

/**
 * Handles a un-subscription request to the current {@link Destination}.
 *
 * @param connection the connection
 * @param frame      the {@code UNSUBSCRIBE} frame
 * @return {@code true} if the un-subscription has been handled, {@code false} otherwise.
 */
@Override
public synchronized boolean unsubscribe(StompServerConnection connection, Frame frame) {
  for (Subscription subscription : new ArrayList<>(subscriptions)) {
    if (subscription.connection.equals(connection)
        && subscription.id.equals(frame.getId())) {

      boolean r = subscriptions.remove(subscription);
      Optional<Subscription> any = subscriptions.stream().filter(s -> s.destination.equals(subscription.destination)).findAny();
      // We unregister the event bus consumer if there are no subscription on this address anymore.
      if (!any.isPresent()) {
        MessageConsumer<?> consumer = registry.remove(subscription.destination);
        if (consumer != null) {
          consumer.unregister();
        }
      }
      return r;
    }
  }
  return false;
}

项目：vertx-stomp 文件：EventBusBridge.java

/**
 * Removes all subscriptions of the given connection
 *
 * @param connection the connection
 * @return the current instance of {@link Destination}
 */
@Override
public synchronized Destination unsubscribeConnection(StompServerConnection connection) {
  new ArrayList<>(subscriptions)
      .stream()
      .filter(subscription -> subscription.connection.equals(connection))
      .forEach(s -> {
        subscriptions.remove(s);
        Optional<Subscription> any = subscriptions.stream().filter(s2 -> s2.destination.equals(s.destination))
            .findAny();
        // We unregister the event bus consumer if there are no subscription on this address anymore.
        if (!any.isPresent()) {
          MessageConsumer<?> consumer = registry.remove(s.destination);
          if (consumer != null) {
            consumer.unregister();
          }
        }
      });
  return this;
}

项目：vertx-service-proxy 文件：ProxyHandler.java

/**
 * Register the proxy handle on the event bus.
 *
 * @param eventBus the event bus
 * @param address the proxy address
 */
public MessageConsumer<JsonObject> register(EventBus eventBus, String address, List<Function<Message<JsonObject>, Future<Message<JsonObject>>>> interceptors) {
  Handler<Message<JsonObject>> handler = this::handle;
  if (interceptors != null) {
    for (Function<Message<JsonObject>, Future<Message<JsonObject>>> interceptor : interceptors) {
      Handler<Message<JsonObject>> prev = handler;
      handler = msg -> {
        Future<Message<JsonObject>> fut = interceptor.apply(msg);
        fut.setHandler(ar -> {
          if (ar.succeeded()) {
            prev.handle(msg);
          } else {
            ReplyException exception = (ReplyException) ar.cause();
            msg.fail(exception.failureCode(), exception.getMessage());
          }
        });
      };
    }
  }
  consumer = eventBus.consumer(address, handler);
  return consumer;
}

项目：he-rss-poll 文件：JDGPutVerticle.java

@Override
public void start() throws Exception {
    super.start();

    MessageConsumer<JsonArray> ebConsumer = vertx.eventBus()
            .consumer(CommonConstants.VERTX_EVENT_BUS_HE_RSS_JDG_PUT);
    ebConsumer.handler(this::processEntries);
}

项目：vxrifa 文件：ReceiverGenerator.java

ReceiverGenerator generateInitializing() {

        tsb = TypeSpec.classBuilder(MessageFormat.format("{0}{1}", interfaceElement.getSimpleName(), VXRIFA_RECEIVER_SUFFIX)).addModifiers(Modifier.PUBLIC);

        tsb.addSuperinterface(ParameterizedTypeName.get(ClassName.get(VxRifaReceiver.class), TypeName.get(interfaceElement.asType())));

        vertxField = FieldSpec.builder(io.vertx.core.Vertx.class, "vertx", Modifier.PRIVATE, Modifier.FINAL).build();
        tsb.addField(vertxField);

        eventBusAddressField = FieldSpec.builder(java.lang.String.class, "eventBusAddress", Modifier.PRIVATE, Modifier.FINAL).build();
        tsb.addField(eventBusAddressField);

        consumersField = FieldSpec.builder(ParameterizedTypeName.get(ClassName.get(List.class), ParameterizedTypeName.get(ClassName.get(MessageConsumer.class), WildcardTypeName.subtypeOf(Object.class))), "consumers", Modifier.PRIVATE)
                .build();        
        tsb.addField(consumersField);

        tsb.addMethod(
                MethodSpec.constructorBuilder()
                        .addModifiers(Modifier.PUBLIC)
                        .addParameter(io.vertx.core.Vertx.class, vertxField.name)
                        .addStatement("this.$N = $N", vertxField, vertxField)
                        .addStatement("this.$N = $S", eventBusAddressField, interfaceElement.getQualifiedName().toString())
                        .build()
        );

        tsb.addMethod(
                MethodSpec.constructorBuilder()
                        .addModifiers(Modifier.PUBLIC)
                        .addParameter(io.vertx.core.Vertx.class, vertxField.name)
                        .addParameter(java.lang.String.class, eventBusAddressField.name)
                        .addStatement("this.$N = $N", vertxField, vertxField)
                        .addStatement("this.$N = $N", eventBusAddressField, eventBusAddressField)
                        .build()
        );

        return this;

    }

项目：ChimeJavaExample 文件：Main.java

static void scheduling(Vertx vertx) {
    EventBus eventBus = vertx.eventBus();
    // Consumer of the timer events
    MessageConsumer<JsonObject> consumer = eventBus.consumer("scheduler:timer");
    // Listens and prints timer events. When timer completes stops the Vertx 
    consumer.handler (
        message -> {
            JsonObject event = message.body();
            if (event.getString("event").equals("complete")) {
                System.out.println("completed");
                vertx.close();
            }
            else {
                System.out.println(event);
            }
        }
    );
    // Create new timer
    eventBus.send (
        "chime",
        (new JsonObject()).put("operation", "create").put("name", "scheduler:timer")
            .put("publish", false).put("max count", 3)
            .put("description", (new JsonObject()).put("type", "interval").put("delay", 1)),
        ar -> {
            if (ar.succeeded()) {
                System.out.println("Scheduling started: " + ar.result().body());
            }
            else {
                System.out.println("Message failed: " + ar.cause());
                vertx.close();
            }
        }
    );
}