Struct ntex_mqtt::v3::MqttServer

pub struct MqttServer<St, H, C, P> { /* private fields */ }

Mqtt v3.1.1 server

St - connection state H - handshake service C - service for handling control messages P - service for handling publish

Every mqtt connection is handled in several steps. First step is handshake. Server calls handshake service with Handshake message, during this step service can authenticate connect packet, it must return instance of connection state St.

Handshake service could be expressed as simple function:

use ntex_mqtt::v3::{Handshake, HandshakeAck};

async fn handshake(hnd: Handshake) -> Result<HandshakeAkc<MyState>, MyError> {
    Ok(hnd.ack(MyState::new(), false))
}

During next stage, control and publish services get constructed, both factories receive Session<St> state object as an argument. Publish service handles Publish packet. On success, server server sends PublishAck packet to the client, in case of error connection get closed. Control service receives all other packets, like Subscribe, Unsubscribe etc. Also control service receives errors from publish service and connection disconnect.

Implementations

impl<St, H> MqttServer<St, H, DefaultControlService<St, H::Error>, DefaultPublishService<St, H::Error>>where
    St: 'static,
    H: ServiceFactory<Handshake, Response = HandshakeAck<St>> + 'static,
    H::Error: Debug,

pub fn new<F>(handshake: F) -> Selfwhere
    F: IntoServiceFactory<H, Handshake>,

Create server factory and provide handshake service

impl<St, H, C, P> MqttServer<St, H, C, P>where
    St: 'static,
    H: ServiceFactory<Handshake, Response = HandshakeAck<St>> + 'static,
    C: ServiceFactory<ControlMessage<H::Error>, Session<St>, Response = ControlResult> + 'static,
    P: ServiceFactory<Publish, Session<St>, Response = ()> + 'static,
    H::Error: From<C::Error> + From<C::InitError> + From<P::Error> + From<P::InitError> + Debug,

pub fn handshake_timeout(self, timeout: Seconds) -> Self

Set handshake timeout.

Handshake includes connect packet and response connect-ack. By default handshake timeuot is disabled.

pub fn disconnect_timeout(self, val: Seconds) -> Self

Set server connection disconnect timeout.

Defines a timeout for disconnect connection. If a disconnect procedure does not complete within this time, the connection get dropped.

To disable timeout set value to 0.

By default disconnect timeout is set to 3 seconds.

pub fn max_qos(self, qos: QoS) -> Self

Set max allowed QoS.

If peer sends publish with higher qos then ProtocolError::MaxQoSViolated(..) By default max qos is set to ExactlyOnce.

pub fn max_size(self, size: u32) -> Self

Set max inbound frame size.

If max size is set to 0, size is unlimited. By default max size is set to 0

pub fn inflight(self, val: u16) -> Self

Number of in-flight concurrent messages.

By default in-flight is set to 16 messages

pub fn inflight_size(self, val: usize) -> Self

Total size of in-flight messages.

By default total in-flight size is set to 64Kb

pub fn control<F, Srv>(self, service: F) -> MqttServer<St, H, Srv, P>where
    F: IntoServiceFactory<Srv, ControlMessage<H::Error>, Session<St>>,
    Srv: ServiceFactory<ControlMessage<H::Error>, Session<St>, Response = ControlResult> + 'static,
    H::Error: From<Srv::Error> + From<Srv::InitError>,

Service to handle control packets

All control packets are processed sequentially, max number of buffered control packets is 16.

pub fn publish<F, Srv>(self, publish: F) -> MqttServer<St, H, C, Srv>where
    F: IntoServiceFactory<Srv, Publish, Session<St>>,
    Srv: ServiceFactory<Publish, Session<St>, Response = ()> + 'static,
    H::Error: From<Srv::Error> + From<Srv::InitError> + Debug,

Set service to handle publish packets and create mqtt server factory

pub fn finish(
    self
) -> MqttServer<Session<St>, impl ServiceFactory<IoBoxed, Response = (IoBoxed, Rc<MqttShared>, Session<St>, Seconds), Error = MqttError<H::Error>, InitError = H::InitError>, impl ServiceFactory<DispatchItem<Rc<MqttShared>>, Session<St>, Response = Option<Packet>, Error = MqttError<H::Error>, InitError = MqttError<H::Error>>, Rc<MqttShared>>

Finish server configuration and create mqtt server factory