Plasma GitLab Archive
Projects Blog Knowledge

Module Netamqp_endpoint

module Netamqp_endpoint: sig .. end
Endpoints are low-level containers for the protocol state

They implement the send/receive queues, and the message dispatcher

type connector = [ `Implied | `Inet of string * int | `Sockaddr of Unix.sockaddr ] 
How to connect to the server:
  • `Sockaddr a: the server port is given by socket address a
  • `Inet(host,port): the server is identified by this host/port pair
  • `Implied: a reserved connector when the server address is implicitly known only

type transport_layer = [ `Custom of
unit -> Netamqp_transport.amqp_multiplex_controller Uq_engines.engine
| `TCP of connector
| `TLS of
connector * (module Netsys_crypto_types.TLS_CONFIG) ]
Which transport layer to choose:
  • `TCP connector: Use TCP
  • `Custom f: Call f () to get a new transport stream

type endpoint 
The abstract endpoint container
type state = [ `Connected of bool
| `Disconnected
| `Disconnecting of bool
| `Error of exn
| `Off ]
  • `Off: Endpoint is still down (initial state)
  • `Connected flag: The endpoint accepts requests. If flag, the TCP connection is established. Note that the endpoint already allows to add requests before the TCP connection is fully established.
  • `Disconnecting flag: A disconnect is requested when all async methods are sent and all sync methods are responded, i.e. when all locally requested operations are done. If flag, the TCP connection is established. The case flag=false can only happen if a not-yet fully connected endpoint is disconnected.
  • `Disconnected: Endpoint is down after regular disconnect
  • `Error: A low-level error occurred, and the endpoint is down

type proto_revision_0_9 = [ `One ] 
type protocol = [ `AMQP_0_9 of proto_revision_0_9 ] 
The protocol variant. Patch releases are not counted here:
  • `AMQP_0_9: This actually chooses AMQP-0-9-1 with as much compatibility as possible for existing servers. 0-9-1 is the subset of 0-9 that is widely implemented.

Methods are control messages exchanged between client and server. They exist in two major styles:

  • Synchronous methods are part of request/response pairs
  • Asynchronous methods are one-way messages
There is a bit confusion in terminology because the sync/async distinction is also used for the way socket events are handled. This is not meant in this context!

Both styles of message exchange can be started by either peer, client or server.

In a request/response pair, each of the two control messages is called a method.

A method usually has arguments. The method type is the method without arguments.

The methods are described in more detail in an XML file (e.g. amqp0-9-1.xml). There is also a version-specific Ocaml module that is generated: Netamqp_method_0_9. See there how the methods are formalized (e.g. which types the arguments have).

type sync_client_to_server_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.sync_client_to_server_method_t ] 
The subset of methods:
  • Synchronous methods the client can send to the server. This includes both initial requests from the client and responses to the server when the server initiates the request/response pair.

type sync_server_to_client_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.sync_server_to_client_method_t ] 
The subset of methods:
  • Synchronous methods the server can send to the client. This includes both initial requests from the server and responses to the client when the client initiates the request/response pair.

type sync_client_initiated_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.sync_client_initiated_method_t ] 
The subset of methods:
  • Synchronous methods where the client sends the first message

type sync_server_initiated_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.sync_server_initiated_method_t ] 
The subset of methods:
  • Synchronous methods where the server sends the first message

type sync_server_initiated_method_type_t = [ `AMQP_0_9 of Netamqp_methods_0_9.sync_server_initiated_method_type_t ] 
The subset of method types:
  • Synchronous method types where the server sends the first message

type async_client_to_server_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.async_client_to_server_method_t ] 
The subset of methods:
  • Asynchronous methods the client can send to the server

type async_server_to_client_method_t = [ `AMQP_0_9 of Netamqp_methods_0_9.async_server_to_client_method_t ] 
The subset of methods:
  • Asynchronous methods the server can send to the server

type async_server_to_client_method_type_t = [ `AMQP_0_9 of Netamqp_methods_0_9.async_server_to_client_method_type_t ] 
The subset of method types:
  • Asynchronous method types the server can send to the server

type method_type_t = [ `AMQP_0_9 of Netamqp_methods_0_9.method_type_t ] 
All method types

Some methods can also carry content data, i.e. the real data that goes to or comes from the message queue. Content data is primarily a long string, but there is also a header with properties.
type props_t = [ `AMQP_0_9 of Netamqp_methods_0_9.props_t ] 
Possible properties. Note that the property class must match the class that is used for content exchange. Right now this means only `P_basic is actually available.
type data = props_t * Netxdr_mstring.mstring list 
Content data as pair of properties and a string. The string is represented as list of mstring, an abstraction over several possible representations of byte arrays provided by Ocamlnet (see the rpc library for Netxdr_mstring).

Data received from the server is often returned as a true list with more than one element. Each element represents a frame on the transport level.

There is no need to split strings into frames before passing the strings to the endpoint for sending them to the server. This is done automatically if needed.

Creating and (dis)conecting

val create : transport_layer ->
protocol ->
Unixqueue.event_system -> endpoint
Creates a new endpoint which is initially `Off.
val configure_timeout : endpoint -> float -> unit
Configures the transport-level timeout. This only affects:
  • Connecting to the server
  • Sending messages
  • Receiving messages while a frame is being read (but not between frames)
Defaults to 300 seconds if not configured.
val get_timeout : endpoint -> float
Return the configured timeout value
val default_port : int
Default port number for AMQP
val connect : endpoint -> unit
Triggers the connect to the server. The new state is `Connected false. If the endpoint is already connected, nothing changes.

After calling connect the underlying transport is still inactive, but the connection procedure is triggered. It is nevertheless immediately possible to add requests to the endpoint. These will be carried out once the connection on the transport level is established.

val disconnect : endpoint -> unit
Requests an orderly disconnect. The disconnect is only triggered but first done when
  • All async methods are sent to the server
  • All locally invoked sync methods are responded
  • The event loop runs the next time.
The new state is `Disconnecting.
val quick_disconnect : endpoint -> unit
A quick disconnect proceeds faster than an orderly disconnect. In particular, it is not waited until the locally invoked sync methods are responded.
val reset : endpoint -> unit
Turns the endpoint off again. Operations, if any, are immediately aborted.
val state : endpoint -> state
Reports the connection state
val state_change_e : endpoint -> state Uq_engines.engine
The engine finishes at the next state change.

It is possible to immediately create another state_change_e when the previous engine finishes. This allows it to continuously watch for state changes.

val event_system : endpoint -> Unixqueue.event_system
Return the event system the endpoint uses
val protocol : endpoint -> protocol
Return the protocol

Using an activated endpoint

The following methods must only be called when the state is `Connected, i.e. after calling connect.
val getsockname : endpoint -> Netamqp_transport.sockaddr
val getpeername : endpoint -> Netamqp_transport.sockaddr
Get the names of the own socket and the peer socket, resp. These functions fail if the endpoint socket is not connected at the moment of checking.
val enable_channel : endpoint -> -> unit
val disable_channel : endpoint -> -> unit
Enable/disable channels. Only an enabled channel can be used for communication.

Channel 0 is always enabled.

Disabling a channel immediately drops all unsent messages except those on the priority queue. Also, pending synchronous calls will get the exception Netamqp_types.Method_dropped. All registrations for the channel are deleted.

val is_channel_enabled : endpoint -> -> bool
Whether a channel is enabled
val suggest_channel : endpoint ->
Suggests a channel number
val flow_control : endpoint -> -> bool -> unit
Requests that the flow is enabled (true) or disabled (false) for a certain channel. By default, the channel flow is enabled.

This only affects content messages sent to the server.

val drop_frames : endpoint -> unit
Request to drop any incoming frames (to be used after having received the connection.close method)
val clear_output : endpoint -> unit
Remove any frames from output queues
val expect_eof : endpoint -> unit
Instruct the endpoint not to generate an Netamqp_types.Unexpected_eof exception when EOF is seen from the server
val set_max_frame_size : endpoint -> int -> unit
val eff_max_frame_size : endpoint -> int
These two function talk to the transport, see Netampq_transport.amqp_multiplex_controller

Sending and receiving messages

val announce_e : endpoint -> unit Uq_engines.engine
val announce_s : endpoint -> unit
Sends the protocol header, and waits until the response arrives. If the response is the right server method, the engine finishes normally. Otherwise it enters the error state.
val sync_c2s_e : ?no_wait:sync_server_to_client_method_t ->
?on_timeout:(unit -> unit) ->
endpoint ->
sync_client_initiated_method_t ->
data option -> ->
float ->
(sync_server_to_client_method_t *
data option)
val sync_c2s_s : ?no_wait:sync_server_to_client_method_t ->
?on_timeout:(unit -> unit) ->
endpoint ->
sync_client_initiated_method_t ->
data option -> ->
float ->
sync_server_to_client_method_t *
data option
Synchronous calls initiated by the client: Calls the argument method and waits for the right reply method.

Only certain methods may be accompanied with a data item (`Basic_return). Only certain methods can have a data item in the response (`Basic_get_ok).

Actually, the possible return methods are much more restricted than sync_server_to_client_method_t, e.g. `Channel_open can only be responded with `Channel_open_ok. This is not reflected in the function type, though.

The float arg is the timeout. If the message is not responded within that time frame, the exception Timeout is raised.

Option no_wait: If set, the function does not wait for the reply, but immediately returns the method no_wait (and no data).

Option on_timeout: This function is called first when a timeout occurs. (Additionally, all pending sync_c2s calls on the same channel get a Timeout exception.)

val register_sync_s2c : endpoint ->
sync_server_initiated_method_type_t -> ->
(sync_server_initiated_method_t ->
sync_client_to_server_method_t option) ->
(unit -> unit) -> unit
Synchronous calls initiated by the server: Registers that the callback function is called when the server sends a method of the given type.

Normally, the callback returns Some r where r is the response method. The callback is also allowed to return None in case of an error. Some additional reaction should be provided, though, e.g. by requesting a connection close.

The post action function is only invoked if the callback returns a result. The idea is that another action can be triggered after the response has been added to the output queue.

The response is not added to the normal output queue, but to the priority output queue, and has precedence to all other methods emitted by the client.

The registered handler is not notified if there is a state change of the endpoint or if an error is propagated. If this is required, the handler should configure additional monitoring for these events.

val async_c2s : endpoint ->
async_client_to_server_method_t ->
data option -> -> unit
Asynchronous calls from the client: This function just sends the given method to the server. Note that the actual transmission is controlled by the event loop and does not happen immediately.

Only certain methods may be accompanied with a data item (`Basic_publish).

There is no indication of any kind whether the operation was successful, not even whether it could be sent to the server. If feedback is required one must use transactions (Tx class).

val async_c2s_e : endpoint ->
async_client_to_server_method_t ->
data option -> -> unit Uq_engines.engine
val async_c2s_s : endpoint ->
async_client_to_server_method_t ->
data option -> -> unit
These versions of async_c2s return first when the message is passed to the socket.
val register_async_s2c : endpoint ->
async_server_to_client_method_type_t -> ->
(async_server_to_client_method_t ->
data option -> unit) ->
Asynchronous calls from the server: Registers that the callback function is called when the server sends a method of the given type.

The registered handler is not notified if there is a state change of the endpoint or if an error is propagated. If this is required, the handler should configure additional monitoring for these events.

Error propagation

Errors from the socket level lead immediately to a shutdown of all activities, and the TCP connection breaks. The state transitions to `Error e, where the exception e describes the error. Some other kinds of locally detected errors are also handled like this. After shutting the endpoint down, the error is propagated (see below).

Exception codes coming from the server are wrapped as `Connection_close or `Channel_close methods. These are forwarded to registered handlers (register_sync_s2c). The handlers can decide to propagate the error code further by calling propagate_error with a Method_exception.

Error propagation is done for all errors detected by the endpoint, but it can also be invoked from outside (function propagate_error, see below).

The following kinds of errors can be generated by the endpoint:

(1) = The endpoint is automatically shut down as described

(2) = This error is not associated to a particular channel

The following codes should be used for injecting errors from outside:

val propagate_error : endpoint -> exn -> option -> unit
Propagate the exception to handlers and engines.

If the channel is None, all possible handlers and engines receive the exception. Otherwise, only the handlers and engines get it that are connected with the given channel.

The engines and handlers receiving the exceptions:

  • announce_e
  • sync_c2s_e
  • The defined listeners (listen_for_errors).
Note that there is no way to notify registered handlers that just wait for incoming server methods (register_sync_s2c and register_async_s2c). If these handlers need to be notified about errors they should register an error listener.

The endpoint state is not modified.

val abort_and_propagate_error : endpoint -> exn -> unit
The endpoint is aborted and set to the `Error state. Also, the passed exception is propagated unconditionally.
val listen_for_errors : endpoint -> option -> (exn -> bool) -> unit
Defines an error listener.

The callback function must return true to remain active; otherwise the listener is disabled.

val create_method_exception : protocol ->
class_id:int -> meth_id:int -> reply_code:int -> reply_text:string -> exn
Returns a Method_exception
val tls_session_props : endpoint -> Nettls_support.tls_session_props option
TLS session properties
module Debug: sig .. end
This web site is published by Informatikbüro Gerd Stolpmann
Powered by Caml