/*
 * Copyright 2014 Red Hat, Inc.
 *
 * Red Hat licenses this file to you under the Apache License, version 2.0
 * (the "License"); you may not use this file except in compliance with the
 * License.  You may obtain a copy of the License at:
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.vertx.reactivex.core.eventbus;

import io.vertx.reactivex.RxHelper;
import io.vertx.reactivex.ObservableHelper;
import io.vertx.reactivex.FlowableHelper;
import io.vertx.reactivex.impl.AsyncResultMaybe;
import io.vertx.reactivex.impl.AsyncResultSingle;
import io.vertx.reactivex.impl.AsyncResultCompletable;
import io.vertx.reactivex.WriteStreamObserver;
import io.vertx.reactivex.WriteStreamSubscriber;
import java.util.Map;
import java.util.Set;
import java.util.List;
import java.util.Iterator;
import java.util.function.Function;
import java.util.stream.Collectors;
import io.vertx.core.Handler;
import io.vertx.core.AsyncResult;
import io.vertx.core.json.JsonObject;
import io.vertx.core.json.JsonArray;
import io.vertx.lang.rx.RxGen;
import io.vertx.lang.rx.TypeArg;
import io.vertx.lang.rx.MappingIterator;

Represents a message that is received from the event bus in a handler.

Messages have a body, which can be null, and also headers, which can be empty.

If the message was sent specifying a reply handler, it can be replied to using reply.

If you want to notify the sender that processing failed, then fail can be called.

NOTE: This class has been automatically generated from the original non RX-ified interface using Vert.x codegen.
/** * Represents a message that is received from the event bus in a handler. * <p> * Messages have a {@link io.vertx.reactivex.core.eventbus.Message#body}, which can be null, and also {@link io.vertx.reactivex.core.eventbus.Message#headers}, which can be empty. * <p> * If the message was sent specifying a reply handler, it can be replied to using {@link io.vertx.reactivex.core.eventbus.Message#reply}. * <p> * If you want to notify the sender that processing failed, then {@link io.vertx.reactivex.core.eventbus.Message#fail} can be called. * * <p/> * NOTE: This class has been automatically generated from the {@link io.vertx.core.eventbus.Message original} non RX-ified interface using Vert.x codegen. */
@RxGen(io.vertx.core.eventbus.Message.class) public class Message<T> { @Override public String toString() { return delegate.toString(); } @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; Message that = (Message) o; return delegate.equals(that.delegate); } @Override public int hashCode() { return delegate.hashCode(); } public static final TypeArg<Message> __TYPE_ARG = new TypeArg<>( obj -> new Message((io.vertx.core.eventbus.Message) obj), Message::getDelegate ); private final io.vertx.core.eventbus.Message<T> delegate; public final TypeArg<T> __typeArg_0; public Message(io.vertx.core.eventbus.Message delegate) { this.delegate = delegate; this.__typeArg_0 = TypeArg.unknown(); } public Message(Object delegate, TypeArg<T> typeArg_0) { this.delegate = (io.vertx.core.eventbus.Message)delegate; this.__typeArg_0 = typeArg_0; } public io.vertx.core.eventbus.Message getDelegate() { return delegate; }
The address the message was sent to
Returns:
/** * The address the message was sent to * @return */
public String address() { String ret = delegate.address(); return ret; }
Multi-map of message headers. Can be empty
Returns:the headers
/** * Multi-map of message headers. Can be empty * @return the headers */
public io.vertx.reactivex.core.MultiMap headers() { io.vertx.reactivex.core.MultiMap ret = io.vertx.reactivex.core.MultiMap.newInstance((io.vertx.core.MultiMap)delegate.headers()); return ret; }
The body of the message. Can be null.
Returns:the body, or null.
/** * The body of the message. Can be null. * @return the body, or null. */
public T body() { if (cached_0 != null) { return cached_0; } T ret = (T)__typeArg_0.wrap(delegate.body()); cached_0 = ret; return ret; }
The reply address. Can be null.
Returns:the reply address, or null, if message was sent without a reply handler.
/** * The reply address. Can be null. * @return the reply address, or null, if message was sent without a reply handler. */
public String replyAddress() { String ret = delegate.replyAddress(); return ret; }
Signals if this message represents a send or publish event.
Returns:true if this is a send.
/** * Signals if this message represents a send or publish event. * @return true if this is a send. */
public boolean isSend() { boolean ret = delegate.isSend(); return ret; }
Reply to this message.

If the message was sent specifying a reply handler, that handler will be called when it has received a reply. If the message wasn't sent specifying a receipt handler this method does nothing.

Params:
  • message – the message to reply with.
/** * Reply to this message. * <p> * If the message was sent specifying a reply handler, that handler will be * called when it has received a reply. If the message wasn't sent specifying a receipt handler * this method does nothing. * @param message the message to reply with. */
public void reply(java.lang.Object message) { delegate.reply(message); }
Link but allows you to specify delivery options for the reply.
Params:
  • message – the reply message
  • options – the delivery options
/** * Link but allows you to specify delivery options for the reply. * @param message the reply message * @param options the delivery options */
public void reply(java.lang.Object message, io.vertx.core.eventbus.DeliveryOptions options) { delegate.reply(message, options); }
Reply to this message, specifying a replyHandler for the reply - i.e. to receive the reply to the reply.

If the message was sent specifying a reply handler, that handler will be called when it has received a reply. If the message wasn't sent specifying a receipt handler this method does nothing.

Params:
  • message – the message to reply with.
  • replyHandler – the reply handler for the reply.
/** * Reply to this message, specifying a <code>replyHandler</code> for the reply - i.e. * to receive the reply to the reply. * <p> * If the message was sent specifying a reply handler, that handler will be * called when it has received a reply. If the message wasn't sent specifying a receipt handler * this method does nothing. * @param message the message to reply with. * @param replyHandler the reply handler for the reply. */
public <R> void replyAndRequest(java.lang.Object message, Handler<AsyncResult<io.vertx.reactivex.core.eventbus.Message<R>>> replyHandler) { delegate.replyAndRequest(message, new Handler<AsyncResult<io.vertx.core.eventbus.Message<R>>>() { public void handle(AsyncResult<io.vertx.core.eventbus.Message<R>> ar) { if (ar.succeeded()) { replyHandler.handle(io.vertx.core.Future.succeededFuture(io.vertx.reactivex.core.eventbus.Message.newInstance((io.vertx.core.eventbus.Message)ar.result(), TypeArg.unknown()))); } else { replyHandler.handle(io.vertx.core.Future.failedFuture(ar.cause())); } } }); }
Reply to this message, specifying a replyHandler for the reply - i.e. to receive the reply to the reply.

If the message was sent specifying a reply handler, that handler will be called when it has received a reply. If the message wasn't sent specifying a receipt handler this method does nothing.

Params:
  • message – the message to reply with.
/** * Reply to this message, specifying a <code>replyHandler</code> for the reply - i.e. * to receive the reply to the reply. * <p> * If the message was sent specifying a reply handler, that handler will be * called when it has received a reply. If the message wasn't sent specifying a receipt handler * this method does nothing. * @param message the message to reply with. */
public <R> void replyAndRequest(java.lang.Object message) { replyAndRequest(message, ar -> { }); }
Reply to this message, specifying a replyHandler for the reply - i.e. to receive the reply to the reply.

If the message was sent specifying a reply handler, that handler will be called when it has received a reply. If the message wasn't sent specifying a receipt handler this method does nothing.

Params:
  • message – the message to reply with.
Returns:
/** * Reply to this message, specifying a <code>replyHandler</code> for the reply - i.e. * to receive the reply to the reply. * <p> * If the message was sent specifying a reply handler, that handler will be * called when it has received a reply. If the message wasn't sent specifying a receipt handler * this method does nothing. * @param message the message to reply with. * @return */
public <R> io.reactivex.Single<io.vertx.reactivex.core.eventbus.Message<R>> rxReplyAndRequest(java.lang.Object message) { return AsyncResultSingle.toSingle($handler -> { replyAndRequest(message, $handler); }); }
Like but specifying options that can be used to configure the delivery.
Params:
  • message – the message body, may be null
  • options – delivery options
  • replyHandler – reply handler will be called when any reply from the recipient is received
/** * Like but specifying <code>options</code> that can be used * to configure the delivery. * @param message the message body, may be <code>null</code> * @param options delivery options * @param replyHandler reply handler will be called when any reply from the recipient is received */
public <R> void replyAndRequest(java.lang.Object message, io.vertx.core.eventbus.DeliveryOptions options, Handler<AsyncResult<io.vertx.reactivex.core.eventbus.Message<R>>> replyHandler) { delegate.replyAndRequest(message, options, new Handler<AsyncResult<io.vertx.core.eventbus.Message<R>>>() { public void handle(AsyncResult<io.vertx.core.eventbus.Message<R>> ar) { if (ar.succeeded()) { replyHandler.handle(io.vertx.core.Future.succeededFuture(io.vertx.reactivex.core.eventbus.Message.newInstance((io.vertx.core.eventbus.Message)ar.result(), TypeArg.unknown()))); } else { replyHandler.handle(io.vertx.core.Future.failedFuture(ar.cause())); } } }); }
Like but specifying options that can be used to configure the delivery.
Params:
  • message – the message body, may be null
  • options – delivery options
/** * Like but specifying <code>options</code> that can be used * to configure the delivery. * @param message the message body, may be <code>null</code> * @param options delivery options */
public <R> void replyAndRequest(java.lang.Object message, io.vertx.core.eventbus.DeliveryOptions options) { replyAndRequest(message, options, ar -> { }); }
Like but specifying options that can be used to configure the delivery.
Params:
  • message – the message body, may be null
  • options – delivery options
Returns:
/** * Like but specifying <code>options</code> that can be used * to configure the delivery. * @param message the message body, may be <code>null</code> * @param options delivery options * @return */
public <R> io.reactivex.Single<io.vertx.reactivex.core.eventbus.Message<R>> rxReplyAndRequest(java.lang.Object message, io.vertx.core.eventbus.DeliveryOptions options) { return AsyncResultSingle.toSingle($handler -> { replyAndRequest(message, options, $handler); }); }
Signal to the sender that processing of this message failed.

If the message was sent specifying a result handler the handler will be called with a failure corresponding to the failure code and message specified here.

Params:
  • failureCode – A failure code to pass back to the sender
  • message – A message to pass back to the sender
/** * Signal to the sender that processing of this message failed. * <p> * If the message was sent specifying a result handler * the handler will be called with a failure corresponding to the failure code and message specified here. * @param failureCode A failure code to pass back to the sender * @param message A message to pass back to the sender */
public void fail(int failureCode, String message) { delegate.fail(failureCode, message); } private T cached_0; public static <T> Message<T> newInstance(io.vertx.core.eventbus.Message arg) { return arg != null ? new Message<T>(arg) : null; } public static <T> Message<T> newInstance(io.vertx.core.eventbus.Message arg, TypeArg<T> __typeArg_T) { return arg != null ? new Message<T>(arg, __typeArg_T) : null; } }