org.apache.beam.sdk.io.rabbitmq

Class RabbitMqIO.Read

java.lang.Object

org.apache.beam.sdk.transforms.PTransform<PBegin,PCollection<RabbitMqMessage>>

org.apache.beam.sdk.io.rabbitmq.RabbitMqIO.Read

All Implemented Interfaces:

java.io.Serializable, HasDisplayData

Enclosing class:

RabbitMqIO

public abstract static class RabbitMqIO.Read
extends PTransform<PBegin,PCollection<RabbitMqMessage>>

A PTransform to consume messages from RabbitMQ server.

See Also:

Serialized Form

Field Summary

Fields inherited from class org.apache.beam.sdk.transforms.PTransform

name, resourceHints

Constructor Summary

Constructors

Constructor and Description
Read()

Method Summary

Modifier and Type	Method and Description
PCollection<RabbitMqMessage>	expand(PBegin input) Override this method to specify how this PTransform should be expanded on the given InputT.
RabbitMqIO.Read	withExchange(java.lang.String name, @Nullable java.lang.String routingKey) In AMQP, messages are published to an exchange and routed to queues based on the exchange type and a queue binding.
RabbitMqIO.Read	withExchange(java.lang.String name, java.lang.String type, @Nullable java.lang.String routingKey) In AMQP, messages are published to an exchange and routed to queues based on the exchange type and a queue binding.
RabbitMqIO.Read	withMaxNumRecords(long maxNumRecords) Define the max number of records received by the RabbitMqIO.Read.
RabbitMqIO.Read	withMaxReadTime(Duration maxReadTime) Define the max read time (duration) while the RabbitMqIO.Read will receive messages.
RabbitMqIO.Read	withQueue(java.lang.String queue) If you want to directly consume messages from a specific queue, you just have to specify the queue name.
RabbitMqIO.Read	withQueueDeclare(boolean queueDeclare) You can "force" the declaration of a queue on the RabbitMQ broker.
RabbitMqIO.Read	withUri(java.lang.String uri)
RabbitMqIO.Read	withUseCorrelationId(boolean useCorrelationId) Toggles deduplication of messages based on the amqp correlation-id property on incoming messages.

Methods inherited from class org.apache.beam.sdk.transforms.PTransform

compose, compose, getAdditionalInputs, getDefaultOutputCoder, getDefaultOutputCoder, getDefaultOutputCoder, getKindString, getName, getResourceHints, populateDisplayData, setResourceHints, toString, validate, validate

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

Read

public Read()

Method Detail

withUri

public RabbitMqIO.Read withUri(java.lang.String uri)

withQueue

public RabbitMqIO.Read withQueue(java.lang.String queue)

If you want to directly consume messages from a specific queue, you just have to specify the queue name. Optionally, you can declare the queue using withQueueDeclare(boolean).

withQueueDeclare

public RabbitMqIO.Read withQueueDeclare(boolean queueDeclare)

You can "force" the declaration of a queue on the RabbitMQ broker. Exchanges and queues are the high-level building blocks of AMQP. These must be "declared" (created) before they can be used. Declaring either type of object ensures that one of that name and of the specified properties exists, creating it if necessary.

NOTE: When declaring a queue or exchange that already exists, the properties specified in the declaration must match those of the existing queue or exchange. That is, if you declare a queue to be non-durable but a durable queue already exists with the same name, the declaration will fail. When declaring a queue, RabbitMqIO will declare it to be non-durable.

Parameters:

queueDeclare - If true, RabbitMqIO will declare a non-durable queue. If another application created the queue, this is not required and should be set to false

withExchange

public RabbitMqIO.Read withExchange(java.lang.String name,
                                    java.lang.String type,
                                    @Nullable java.lang.String routingKey)

In AMQP, messages are published to an exchange and routed to queues based on the exchange type and a queue binding. Most exchange types utilize the routingKey to determine which queues to deliver messages to. It is incumbent upon the developer to understand the paradigm in place to determine whether to declare a queue, what the appropriate binding should be, and what routingKey will be in use.

This function should be used if the Beam pipeline will be responsible for declaring the exchange. As a result of calling this function, exchangeDeclare will be set to true and the resulting exchange will be non-durable and of the supplied type. If an exchange with the given name already exists but is durable or is of another type, exchange declaration will fail.

To use an exchange without declaring it, especially for cases when the exchange is shared with other applications or already exists, use withExchange(String, String) instead.

See Also:

"https://www.cloudamqp.com/blog/2015-09-03-part4-rabbitmq-for-beginners-exchanges-routing-keys-bindings.html" for a write-up on exchange types and routing semantics

withExchange

public RabbitMqIO.Read withExchange(java.lang.String name,
                                    @Nullable java.lang.String routingKey)

In AMQP, messages are published to an exchange and routed to queues based on the exchange type and a queue binding. Most exchange types utilize the routingKey to determine which queues to deliver messages to. It is incumbent upon the developer to understand the paradigm in place to determine whether to declare a queue, with the appropriate binding should be, and what routingKey will be in use.

This function should be used if the Beam pipeline will be using an exchange that has already been declared or when using an exchange shared by other applications, such as an events bus or pubsub. As a result of calling this function, exchangeDeclare will be set to false.

withMaxNumRecords

public RabbitMqIO.Read withMaxNumRecords(long maxNumRecords)

Define the max number of records received by the RabbitMqIO.Read. When this max number of records is lower than Long.MAX_VALUE, the RabbitMqIO.Read will provide a bounded PCollection.

withMaxReadTime

public RabbitMqIO.Read withMaxReadTime(Duration maxReadTime)

Define the max read time (duration) while the RabbitMqIO.Read will receive messages. When this max read time is not null, the RabbitMqIO.Read will provide a bounded PCollection.

withUseCorrelationId

public RabbitMqIO.Read withUseCorrelationId(boolean useCorrelationId)

Toggles deduplication of messages based on the amqp correlation-id property on incoming messages.

When set to true all read messages will require the amqp correlation-id property to be set.

When set to false the correlation-id property will not be used by the Reader and no automatic deduplication will occur.

expand

public PCollection<RabbitMqMessage> expand(PBegin input)

Description copied from class: PTransform

Override this method to specify how this PTransform should be expanded on the given InputT.

NOTE: This method should not be called directly. Instead apply the PTransform should be applied to the InputT using the apply method.

Composite transforms, which are defined in terms of other transforms, should return the output of one of the composed transforms. Non-composite transforms, which do not apply any transforms internally, should return a new unbound output and register evaluators (via backend-specific registration methods).

Specified by:

expand in class PTransform<PBegin,PCollection<RabbitMqMessage>>