SYNOPSIS

my $client = Net::MQTT.new(:server<localhost>);
react {
	whenever $client.subscribe('raku/updates/#') -> $update {
		say "{$update.topic}: {$update.payload.decode('utf8-c8')}";
	}
	whenever Supply.interval(20) {
		$client.publish('time/now', "Current time is {Datetime.now}", :retain);
	}
	whenever $client.connected {
		say "Connected";
		LAST { done }
	}
}

DESCRIPTION

Protocol::MQTT::Client is a network and time independent implementation of a MQTT client.

METHODS

new(*%args)

This creates a new mqtt client. It takes the follow arguments, all are optional:

• Str :$client-identifier

  An unique identifier for this client/session. If none is given a random identifier will be generated.

• Int :$keep-alive-interval

  This sets the keep-alive interval (in seconds). This defaults to 60.

• Int :$resend-interval

  This sets the resend interval (in seconds). This defaults to 5.

• Int :$connect-interval

  This sets the connect interval. It defaults to the resend interval.

• Int :$reconnect-attempts

  This sets how many times it will try to (re)connect before giving up. This defaults to 3.

• Str :$username

  This sets the username, if any.

• Blob :$password

  This sets the password, if any.

• Protocol::MQTT::Message :$will

  This is the will of the connection. If the connection is lost without a formal disconnect, the server will send this message as if the client sent it.

next-events(Instant $now -> List:D)

This returns a list of packets to be sent over the connection. This must be called after any method that changes (connect, disconnect, publish, subscribe, unsubscribe, received-packet), or when next-expiration times out.

connect()

This changes the state to connecting.

disconnect(--> Nil)

This marks the connection for disconnecting.

publish(Str:D $topic, Blob:D $payload, Qos:D $quality-of-service, Bool:D $retainedness, Instant:D $now --> Promise:D)

This publishes $payload to $topic, with the specified $quality-of-service and $retainedness. The resulting promise will succeed when the QOS requirements are met (or fail if the connection is lost before that happens).

subscribe(Str:D $topic, Qos:D $qos, Instant:D $now --> Promise:D)

This subscribes to a specific topic with messages being sent using the specified quality-of-service. The messages will be posted on the incoming Supply.

unsubscribe(Str:D $topic, Instant:D $now --> Promise:D)

This unsubscribes from a $topic.

received-packet(Protocol::MQTT::Packet $packet, Instant $now)

This should be called with every incoming Packet. Turning a data-stream into packets is done using Protocol::MQTT::PacketBuffer.

next-expiration(--> Instant:D)

This returns the next Instant when current outstanding message expire and hence next-events should be called again, if any. This is typically called right after next-events is called.

incoming(--> Supply:D)

This returns the supply of Protocol::MQTT::Message objects that have been received from the server.

disconnected(--> Promise:D)

This promise will succeed whenever the connection is lost (connecting failed, or we failed to receive keep-alives for too long).