Module turtle

Description
Data Types
Function Index
Function Details

External API for the Turtle subsystem.

Description

External API for the Turtle subsystem.

The interface to the Turtle system for other subsystems. You probably want to start a turtle_publisher and a turtle_subscriber in your application, and then use the publish/5 command in this module to publish messages. The subscriber is used by providing a function to it, which is called whenever messages arrive on the subscribed channel.

Other functions in this module are low-level functions. They are not meant for use in applications directly, but in the future, we may have to tip-toe ourselves through a special case or two. Hence, they are still exported and documented so people know they are there.

Data Types

rpc_tag()

rpc_tag() = {pid(), integer()}

time()

time() = integer()

Function Index

await/2
await/3
declare/3
i/0	i/0 returns information about turtle children in the system.
i/1
publish/5	Equivalent to `publish(Name, Exch, Key, ContentType, Payload, #{delivery_mode => ephemeral})`.
publish/6	publish(Name, Exch, Key, ContentType, Payload, Opts) publishes messages.
publish_sync/5	Equivalent to `publish_sync(Name, Exch, Key, ContentType, Payload, #{delivery_mode => ephemeral})`.
publish_sync/6	publish_sync/6 publishes messages synchronously.
rpc/5	rpc/5 performs RPC calls over a publisher.
rpc_await/3	rpc_await/3 awaits the response of an opaque value.
rpc_await_monitor/3	rpc_await_monitor/3 awaits a response or a monitor timeout.
rpc_cancel/2	rpc_cancel/2 cancels an opaque message on the publisher.
rpc_sync/5	Equivalent to `rpc_sync(Pub, X, Key, CType, Payload, #{timeout => 5000})`.
rpc_sync/6	rpc_sync/6 performs a synchronous RPC call over AMQP.

Function Details

await/2

await(X1, Name) -> any()

await/3

await(X1, Name, Timeout) -> any()

declare/3

declare(Channel, Ds, Opts) -> any()

i/0

i() -> #{atom() => any()}

i/0 returns information about turtle children in the system

i/1

i(X1::publishers) -> #{any() => pid()}

publish/5

publish(Pub, Exchange, Key, CType, Payload) -> ok

Pub = atom()
Exchange = binary()
Key = binary()
CType = binary()
Payload = iodata()

Equivalent to publish(Name, Exch, Key, ContentType, Payload, #{delivery_mode => ephemeral}).

publish/6

publish(Pub, Exchange, Key, CType, Payload, Opts) -> ok

Pub = atom()
Exchange = binary()
Key = binary()
CType = binary()
Payload = iodata()
Opts = #{atom() => any()}

publish(Name, Exch, Key, ContentType, Payload, Opts) publishes messages.

This publication variant requires you to have started a publisher already through the supervisor tree. It will look up the appropriate publisher under the given Name, and will publish on Exch with routing key Key, content-type ContentType and the given Payload.

Options is a map of options. Currently we support:

delivery_mode :: persistent | ephemeral (ephemeral is the default)

Publication is asynchronous, so it never fails, but of course, if network conditions are outright miserable, it may fail to publish the message.

publish_sync/5

publish_sync(Pub, Exchange, Key, CType, Payload) -> {ack | nack, time()}

Pub = atom()
Exchange = binary()
Key = binary()
CType = binary()
Payload = iodata()

Equivalent to publish_sync(Name, Exch, Key, ContentType, Payload, #{delivery_mode => ephemeral}).

publish_sync/6

publish_sync(Pub, Exchange, Key, CType, Payload, Opts) -> {ack | nack, time()}

Pub = atom()
Exchange = binary()
Key = binary()
CType = binary()
Payload = iodata()
Opts = #{atom() => any()}

publish_sync/6 publishes messages synchronously

This variant of publish, will publish the message synchronously to the broker, and wait for the best effort of delivery guarantee. Without publisher confirms, this is the point where the message is delivered to the TCP network stack. With confirms, the call will block until the Broker either acks or nacks it.

rpc/5

rpc(Pub, Exch, Key, CType, Payload) -> {ok, Opaque, T}

Pub = atom()
Exch = binary()
Key = binary()
CType = binary()
Payload = iodata()
Opaque = rpc_tag()
T = time()

rpc/5 performs RPC calls over a publisher

The call returns {ok, Opaque, T} where Opaque is an opaque token for the query, and T is the time it took for confirmation in milli seconds.

rpc_await/3

rpc_await(Pub, Opaque, Timeout) -> {ok, T, CType, Payload} | {error, Reason}

Pub = atom()
Opaque = rpc_tag()
Timeout = non_neg_integer()
T = time()
CType = binary()
Payload = binary()
Reason = term()

rpc_await/3 awaits the response of an opaque value

rpc_await_monitor/3

rpc_await_monitor(Opaque, Timeout, MRef) -> {ok, T, CType, Payload} | {error, Reason}

Opaque = rpc_tag()
Timeout = non_neg_integer()
MRef = reference()
T = time()
CType = binary()
Payload = binary()
Reason = term()

rpc_await_monitor/3 awaits a response or a monitor timeout

This variant allows you to reuse a monitor rather than setting a new one every time on the publisher. One tends to be enough :)

rpc_cancel/2

rpc_cancel(Publisher, Opaque) -> ok

Publisher = atom()
Opaque = rpc_tag()

rpc_cancel/2 cancels an opaque message on the publisher

rpc_sync/5

rpc_sync(Pub, X, Key, CType, Payload) -> {ok, T, RCType, RPayload} | {error, Reason}

Pub = atom()
X = binary()
Key = binary()
CType = binary()
Payload = iodata()
T = time()
RCType = binary()
RPayload = binary()
Reason = term()

Equivalent to rpc_sync(Pub, X, Key, CType, Payload, #{timeout => 5000}).

rpc_sync/6

rpc_sync(Pub, X, Key, CType, Payload, Opts) -> {ok, T, RCType, RPayload} | {error, Reason}

Pub = any()
X = binary()
Key = binary()
CType = binary()
Payload = iodata()
Opts = #{atom() => any()}
T = time()
RCType = binary()
RPayload = binary()
Reason = term()

rpc_sync/6 performs a synchronous RPC call over AMQP

Generated by EDoc