/*
* @Copyright (c) Huawei Technologies Co., Ltd. 2023-2024. All rights reserved.
*/
package mqtt4cj.mqttv3
/**
* Bridge between Receiver and the external API.
* This class gets called by Receiver,
* and then converts the comms-centric MQTT message objects
* Int64o ones understood by the external API.
*/
class CommsCallback {
private static let CLASS_NAME: String = "CommsCallback"
private static let INBOUND_QUEUE_SIZE: Int64 = 10
private let log: IMqttLogger = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
private let lifecycle: Mutex = Mutex()
private let workAvailable: Mutex = Mutex()
private let workAvailableCondition = synchronized(workAvailable){workAvailable.condition()}
private let spaceAvailable: Mutex = Mutex()
private let spaceAvailableCondition = synchronized(spaceAvailable){spaceAvailable.condition()}
private var current_state: CallbackState = CallbackState.STOPPED
private var target_state: CallbackState = CallbackState.STOPPED
private var manualAcks: Bool = false
private var threadName: String = CLASS_NAME
private let clientComms: ClientComms
private let callbacksWildcards: ConcurrentHashMap<String, IMqttMessageListener> // topicFilter with wildcards -> messageHandler
private let callbacksDirect: ConcurrentHashMap<String, IMqttMessageListener> // topicFilter without wildcards -> messageHandler
private let messageQueue: ConcurrentLinkedQueue<MqttWireMessage>
private let completeQueue: ConcurrentLinkedQueue<MqttToken>
private let messageQueueLock = Mutex()
private let completeQueueLock = Mutex()
private var mqttCallback: ?MqttCallback = None
private var reconnectInternalCallback: ?MqttCallbackExtended = None
private var callbackThread: ?Thread = None
private var callbackFuture: ?Future<Unit> = None
private var clientState: ?ClientState = None
init(clientComms: ClientComms) {
this.clientComms = clientComms
this.messageQueue = ConcurrentLinkedQueue<MqttWireMessage>()
this.completeQueue = ConcurrentLinkedQueue<MqttToken>()
this.callbacksWildcards = ConcurrentHashMap<String, IMqttMessageListener>()
this.callbacksDirect = ConcurrentHashMap<String, IMqttMessageListener>()
log.setResourceName(clientComms.getClient().getClientId())
}
public func setClientState(clientState: ClientState): Unit {
this.clientState = clientState
}
/**
* Starts up the Callback thread.
*
* @param threadName The name of the thread
*/
public func start(threadName: String): Unit {
this.threadName = threadName
synchronized(lifecycle) {
if (current_state == CallbackState.STOPPED) {
// Preparatory work before starting the background thread.
// For safety ensure any old events are cleared.
messageQueue.clear()
completeQueue.clear()
target_state = CallbackState.RUNNING
current_state = CallbackState.RUNNING
let future = spawn {=> this.run()}
callbackFuture = future
callbackThread = future.thread
}
}
let stoppedStateCounter: AtomicInt64 = AtomicInt64(0)
while (!isRunning()) {
sleep(Duration.millisecond * 100)
if (current_state == CallbackState.STOPPED) {
if (stoppedStateCounter.fetchAdd(1) + 1 > CommsSender.MAX_STOPPED_STATE_TO_STOP_THREAD) {
break
}
} else {
stoppedStateCounter.store(0)
}
}
}
/**
* Stops the callback thread.
* This call will block until stop has completed.
*/
public func stop(): Unit {
let methodName: String = "stop"
if (isRunning()) {
// @TRACE 700=stopping
log.trace(CLASS_NAME, methodName, "700")
synchronized(lifecycle) {
target_state = CallbackState.STOPPED
}
// Do not allow a thread to wait for itself.
if (Thread.currentThread != callbackThread()) {
synchronized(workAvailable) {
// @TRACE 701=notify workAvailable and wait for run
// to finish
log.trace(CLASS_NAME, methodName, "701")
workAvailableCondition.notifyAll()
}
// Wait for the thread to finish.
if (callbackFuture.isSome() && callbackFuture().thread != Thread.currentThread) {
try {
callbackFuture?.cancel()
callbackFuture?.get(SHUTDOWN_TIMEOUT)
} catch (_) {
}
}
}
// @TRACE 703=stopped
log.trace(CLASS_NAME, methodName, "703")
}
}
public func setCallback(mqttCallback: MqttCallback): Unit {
this.mqttCallback = mqttCallback
}
public func setReconnectCallback(callback: MqttCallbackExtended): Unit {
this.reconnectInternalCallback = callback
}
public func setManualAcks(manualAcks: Bool): Unit {
this.manualAcks = manualAcks
}
public func run(): Unit {
let methodName: String = "run"
Thread.currentThread.name = threadName
callbackThread = Thread.currentThread
while (isRunning()) {
try {
// If no work is currently available, then wait until there is some...
synchronized(workAvailable) {
if (isRunning() && messageQueue.size == 0 && completeQueue.size == 0) {
// @TRACE 704=wait for workAvailable
log.trace(CLASS_NAME, methodName, "704")
workAvailableCondition.wait()
}
}
if (isRunning()) {
// Check for deliveryComplete callbacks...
var token: ?MqttToken = None
synchronized(completeQueueLock) {
if (completeQueue.size > 0) {
// First call the delivery arrived callback if needed
token = completeQueue.remove()
}
}
if (token.isSome()) {
handleActionComplete(token())
}
// Check for messageArrived callbacks...
var message: ?MqttPublish = None
synchronized(messageQueueLock) {
if (messageQueue.size > 0) {
// Note, there is a window on connect where a publish
// could arrive before we've finished the connect logic.
message = match (messageQueue.remove()) {
case Some(v: MqttPublish) => v
case _ => None
}
}
}
if (message.isSome()) {
handleMessage(message())
}
}
if (isQuiescing()) {
clientState?.checkQuiesceLock()
}
} catch (ex: Exception) {
// Users code could throw an Error or Exception e.g. in the case
// of class NoClassDefFoundError
// @TRACE 714=callback threw exception
log.trace(CLASS_NAME, methodName, "714", None, ex)
clientComms.shutdownConnection(None, MqttException(ex))
} finally {
synchronized(spaceAvailable) {
// Notify the spaceAvailable lock, to say that there's now
// some space on the queue...
// @TRACE 706=notify spaceAvailable
log.trace(CLASS_NAME, methodName, "706")
spaceAvailableCondition.notifyAll()
}
}
}
synchronized(lifecycle) {
current_state = CallbackState.STOPPED
}
}
private func handleActionComplete(token: MqttToken): Unit {
let methodName: String = "handleActionComplete"
synchronized(token.monitor) {
// @TRACE 705=callback and notify for key={0}
log.trace(CLASS_NAME, methodName, "705", [token.internalTok.getKey()])
if (token.isComplete()) {
// Finish by doing any post processing such as delete
// from persistent store but only do so if the action
// is complete
clientState?.notifyComplete(token)
}
// Unblock any waiters and if pending complete now set completed
token.internalTok.notifyComplete()
if (!token.internalTok.isNotified()) {
// If a callback is registered and delivery has finished
// call delivery complete callback.
if (mqttCallback.isSome() && token is MqttDeliveryToken && token.isComplete()) {
mqttCallback?.deliveryComplete((token as MqttDeliveryToken)())
}
// Now call async action completion callbacks
fireActionEvent(token)
}
// Set notified so we don't tell the user again about this action.
if (token.isComplete()) {
if (token is MqttDeliveryToken) {
token.internalTok.setNotified(true)
}
}
}
}
/**
* This method is called when the connection to the server is lost.
* If there is no cause then it was a clean disconnect.
* The connectionLost callback will be invoked if registered and run on
* the thread that requested shutdown e.g. receiver or sender thread.
* If the request was a user initiated disconnect then the disconnect token will be notified.
*
* @param cause the reason behind the loss of connection.
*/
public func connectionLost(cause: ?MqttException): Unit {
let methodName: String = "connectionLost"
// If there was a problem and a client callback has been set inform
// the connection lost listener of the problem.
try {
if (mqttCallback.isSome() && cause.isSome()) {
// @TRACE 708=call connectionLost
log.trace(CLASS_NAME, methodName, "708", [cause])
mqttCallback?.connectionLost(cause())
}
if (reconnectInternalCallback.isSome() && cause.isSome()) {
reconnectInternalCallback?.connectionLost(cause())
}
} catch (t: Exception) {
// Just log the fact that a throwable has caught connection lost
// is called during shutdown processing so no need to do anything else
// @TRACE 720=exception from connectionLost {0}
log.trace(CLASS_NAME, methodName, "720", [t])
}
}
/**
* An action has completed - if a completion listener has been set on the
* token then invoke it with the outcome of the action.
*
* @param token The MqttToken that has completed
*/
public func fireActionEvent(token: MqttToken): Unit {
let methodName: String = "fireActionEvent"
let asyncCB: ?IMqttActionListener = token.getActionCallback()
if (asyncCB.isSome()) {
if (token.getException().isNone()) {
// @TRACE 716=call onSuccess key={0}
log.trace(CLASS_NAME, methodName, "716", [token.internalTok.getKey()])
asyncCB?.onSuccess(token)
} else {
// @TRACE 717=call onFailure key {0}
log.trace(CLASS_NAME, methodName, "716", [token.internalTok.getKey()])
asyncCB?.onFailure(token, token.getException()())
}
}
}
/**
* This method is called when a message arrives on a topic.
* Messages are only added to the queue for inbound messages if the client is not quiescing.
*
* @param sendMessage the MQTT SEND message.
*/
public func messageArrived(sendMessage: MqttPublish): Unit {
let methodName: String = "messageArrived"
if (mqttCallback.isSome() || !callbacksWildcards.isEmpty() || !callbacksDirect.isEmpty()) {
// If we already have enough messages queued up in memory,
// wait until some more queue space becomes available.
// This helps the client protect itself from getting flooded by messages from the server.
synchronized(spaceAvailable) {
while (isRunning() && !isQuiescing() && messageQueue.size >= INBOUND_QUEUE_SIZE) { /*cjlint-ignore !G.EXP.03 */
try {
// @TRACE 709=wait for spaceAvailable
log.trace(CLASS_NAME, methodName, "709")
spaceAvailableCondition.wait(timeout: Duration.millisecond * 200)
} catch (ex: Exception) {
}
}
}
if (!isQuiescing()) {
messageQueue.add(sendMessage)
// Notify the CommsCallback thread that there's work to do...
synchronized(workAvailable) {
// @TRACE 710= msg avail, notify workAvailable
log.trace(CLASS_NAME, methodName, "710")
workAvailableCondition.notifyAll()
}
}
}
}
/**
* Let the call back thread quiesce.
* Prevent inbound messages being added to the process queue and let existing work quiesce.
* (until the thread is told to shutdown).
*/
public func quiesce(): Unit {
let methodName: String = "quiesce"
synchronized(lifecycle) {
if (current_state == CallbackState.RUNNING) {
current_state = CallbackState.QUIESCING
}
}
synchronized(spaceAvailable) {
// @TRACE 711=quiesce notify spaceAvailable
log.trace(CLASS_NAME, methodName, "711")
// Unblock anything waiting for space...
spaceAvailableCondition.notifyAll()
}
}
public func isQuiesced(): Bool {
return isQuiescing() && completeQueue.size == 0 && messageQueue.size == 0
}
private func handleMessage(publishMessage: MqttPublish): Unit {
let methodName: String = "handleMessage"
// If quisecing process any pending messages.
let destName: String = publishMessage.getTopicName()
// @TRACE 713=call messageArrived key={0} topic={1}
log.trace(CLASS_NAME, methodName, "713", [publishMessage.getMessageId(), destName])
deliverMessage(destName, publishMessage.getMessageId(), publishMessage.getMessage())
if (!this.manualAcks) {
if (publishMessage.getMessage().getQos() == 1) {
this.clientComms.internalSend(
MqttPubAck(publishMessage),
MqttToken(clientComms.getClient().getClientId())
)
} else if (publishMessage.getMessage().getQos() == 2) {
this.clientComms.deliveryComplete(publishMessage)
let pubComp: MqttPubComp = MqttPubComp(publishMessage)
this.clientComms.internalSend(pubComp, MqttToken(clientComms.getClient().getClientId()))
}
}
}
public func messageArrivedComplete(messageId: Int64, qos: Int64): Unit {
if (qos == 1) {
this.clientComms.internalSend(MqttPubAck(messageId), MqttToken(clientComms.getClient().getClientId()))
} else if (qos == 2) {
this.clientComms.deliveryComplete(messageId)
let pubComp: MqttPubComp = MqttPubComp(messageId)
this.clientComms.internalSend(pubComp, MqttToken(clientComms.getClient().getClientId()))
}
}
public func asyncOperationComplete(token: MqttToken): Unit {
let methodName: String = "asyncOperationComplete"
if (isRunning()) {
// invoke callbacks on callback thread
completeQueue.add(token)
synchronized(workAvailable) {
// @TRACE 715= workAvailable. key={0}
log.trace(CLASS_NAME, methodName, "715", [token.internalTok.getKey()])
workAvailableCondition.notifyAll()
}
} else {
// invoke async callback on invokers thread
try {
handleActionComplete(token)
} catch (ex: Exception) {
// Users code could throw an Error or Exception e.g. in the case
// of class NoClassDefFoundError
// @TRACE 719=callback threw ex:
log.trace(CLASS_NAME, methodName, "719", None, ex)
// Shutdown likely already in progress but no harm to confirm
clientComms.shutdownConnection(None, MqttException(ex))
}
}
}
/**
* Returns the thread used by this callback.
*
* @return The Thread
*/
protected func getThread(): ?Thread {
return callbackThread
}
public func setMessageListener(topicFilter: String, messageListener: IMqttMessageListener): Unit {
if (topicFilter.contains("#") || topicFilter.contains("+")) {
this.callbacksWildcards.add(topicFilter, messageListener)
} else {
this.callbacksDirect.add(topicFilter, messageListener)
}
}
public func removeMessageListener(topicFilter: String): Unit {
this.callbacksWildcards.remove(topicFilter) // no exception thrown if the filter was not present
this.callbacksDirect.remove(topicFilter) // no exception thrown if the filter was not present
}
public func removeMessageListeners(): Unit {
this.callbacksWildcards.clear()
this.callbacksDirect.clear()
}
protected func deliverMessage(topicName: String, messageId: Int64, aMessage: MqttMessage): Bool {
var delivered: Bool = false
let callback: ?IMqttMessageListener = this.callbacksDirect.get(topicName)
if (callback.isSome()) {
aMessage.setId(messageId)
callback?.messageArrived(topicName, aMessage)
delivered = true
}
for ((topicFilter, callback) in callbacksWildcards) {
// callback may already have been removed in the meantime, so a None check is necessary
if (MqttTopic.isMatched(topicFilter, topicName)) {
aMessage.setId(messageId)
callback.messageArrived(topicName, aMessage)
delivered = true
}
}
/* if the message hasn't been delivered to a per subscription handler, give it to the default handler */
if (mqttCallback.isSome() && !delivered) {
aMessage.setId(messageId)
mqttCallback?.messageArrived(topicName, aMessage)
delivered = true
}
return delivered
}
public func isRunning(): Bool {
let result: Bool
synchronized(lifecycle) {
result = ((current_state == CallbackState.RUNNING || current_state == CallbackState.QUIESCING) &&
target_state == CallbackState.RUNNING)
}
return result
}
public func isQuiescing(): Bool {
let result: Bool
synchronized(lifecycle) {
result = (current_state == CallbackState.QUIESCING)
}
return result
}
}
enum CallbackState {
| STOPPED
| RUNNING
| QUIESCING
operator func ==(that: CallbackState): Bool {
match ((this, that)) {
case (STOPPED, STOPPED) => true
case (RUNNING, RUNNING) => true
case (QUIESCING, QUIESCING) => true
case _ => false
}
}
}
public interface IDiscardedBufferMessageCallback {
func messageDiscarded(message: MqttWireMessage): Unit
}
public interface IDisconnectedBufferCallback {
func publishBufferedMessage(bufferedMessage: BufferedMessage): Unit
}
