/*
* @Copyright (c) Huawei Technologies Co., Ltd. 2023-2024. All rights reserved.
*/
package mqtt4cj.mqttv3
/**
A NetworkModule provides access to a specific transport to the broker.
This may be a plain socket, a TLS secured connection, a serial line, etc.
Lifecycle
Each NetworkModule instance has a lifecycle with the following logical states:
CREATED - the instance holds no connection to its broker, no network resources are allocated and the get...Stream-methods may return null or throw an IOException.
CONNECTED - the instance has an active connection to its broker, underlaying network ressources (e.g. socket) are allocated, the get...Stream-methods return open streams for reading and writing.
DISCONNECTED - the instance holds no connection to its broker, underlaying network resources are closed and released, the get...Stream-methods will always throw an IOException.
The following transitions may occure:
CREATED -> CONNECTED - if start() succeeded
CREATED -> CREATED - if start() failed
CONNECTED -> DISCONNECTED - when stop() is called; If an IOException occures on one of the input / output streams the stop()-method will be called.
DISCONNECTED -> CONNECTED - if start() succeeded during a re-connect attempt
The start()-method of a NetworkModule instance may only be called multiple times if either the previous call resulted in an exception or stop() has been called in between.
The objects returned by getInputStream() und getOutputStream() are most likely different instances each time start() had been called.
*/
public interface NetworkModule {
/**
* Open the transport to the broker.
* The streams provided by getInputStream() and getOutputStream() are expected to be open
* for read / write operations after this method succeed.
*
* @throws IOException of any underlaying transport
* @throws MqttException if the server connection cannot be established, e.g. the connection is being refused
*/
func start(): Unit
/**
* Returns the input stream to be used for receiving messages.
* This method is usually called once directly after start().
* The returned input stream will be used in the CommsReceiver of the client connection.
*
* @return the input stream to be used for receiving messages (null may be returned if start() had never been sucessfully called on this instance)
* @throws IOException – if an I/O error occurs when creating the input stream or the broker connection is not established
*/
func getInputStream(): InputStream
/**
* Returns the output stream to be used for sending messages.
* This method is usually called once directly after start().
* The returned output stream will be used in the CommsSender of the client connection.
*
* @return output stream to be used for sending messages (null may be returned if start() had never been sucessfully called on this instance)
* @throws IOException – if an I/O error occurs when creating the output stream or the broker connection is not established
*/
func getOutputStream(): OutputStream
/**
* Close the transport to the broker.
* The streams provided by getInputStream() and getOutputStream()
* should be closed after this method returns.
*
* @throws IOException if an I/O error occurs when closing the transport
*/
func stop(): Unit
/**
* Returns the URI of the broker which was used to create this NetworkModule.
*
* @return the URI of the broker
*/
func getServerURI(): String
}
public type SocketFactory = (String) -> TcpSocket
let DEFAULT_SOCKET_FACTORY: SocketFactory = {
urlStr: String =>
LOGGER.trace("", "DEFAULT_SOCKET_FACTORY", urlStr)
let url = URL.parse(urlStr)
TcpSocket(url.hostName, UInt16.parse(url.port))
}
let DEFAULT_TLS_CLIENT_CONFIG: TlsClientConfig = {
=>
var cfg = TlsClientConfig()
cfg.verifyMode = TrustAll
cfg
}()
/**
* A network module for connecting over TCP.
*/
public class TCPNetworkModule <: NetworkModule {
private static let CLASS_NAME = "TCPNetworkModule"
private let log = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
protected var socket: ?TcpSocket = None
private let factory: SocketFactory
private let brokerUrl: URL
private let option: MqttConnectOptions
/**
* Constructs a new TCPNetworkModule using the specified host and
* port. The supplied SocketFactory is used to supply the network
* socket.
* @param brokerUrl - URL
* @param resourceContext - String
* @param factory - SocketFactory
* @param connTimeout - ?Duration
*/
public init(
brokerUrl: URL,
resourceContext: String,
factory!: SocketFactory,
option!: MqttConnectOptions
) {
log.setResourceName(resourceContext)
this.factory = factory
this.brokerUrl = brokerUrl
this.option = option
}
/**
* Starts the module, by creating a TCP socket to the server.
* @throws IOException if there is an error creating the socket
* @throws MqttException if there is an error connecting to the server
*/
public func start(): Unit {
let methodName: String = "start"
try {
// @TRACE 252=connect to host {0} port {1} timeout {2}
log.trace(CLASS_NAME, methodName, "252", [brokerUrl.hostName, brokerUrl.port, option.getConnectionTimeout()])
this.socket = this.factory(this.getServerURI())
this.socket?.connect(timeout: option.getConnectionTimeout())
this.socket?.readTimeout = option.getKeepAliveInterval() * 2
} catch (ex: SocketException | SocketTimeoutException | IllegalArgumentException) {
// @TRACE 250=Failed to create TCP socket
log.trace(CLASS_NAME, methodName, "250", None, ex)
throw MqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR, ex)
}
}
/**
* Returns the input stream to be used for receiving messages. This method is usually called once directly after
* {@link #start()}. The returned input stream will be used in the CommsReceiver of the client connection.
*
* @return the input stream to be used for receiving messages ({@code null} may be returned if {@link #start()} had
* never been sucessfully called on this instance)
* @throws IOException if an I/O error occurs when creating the input stream or the broker connection is not
* established
*/
public func getInputStream(): InputStream {
return socket()
}
/**
* Returns the output stream to be used for sending messages. This method is usually called once directly after
* {@link #start()}. The returned output stream will be used in the CommsSender of the client connection.
*
* @return output stream to be used for sending messages ({@code null} may be returned if {@link #start()} had never
* been sucessfully called on this instance)
* @throws IOException if an I/O error occurs when creating the output stream or the broker connection is not
* established
*/
public func getOutputStream(): OutputStream {
return socket()
}
/**
* Close the transport to the broker. The streams provided by {@link #getInputStream()} and
* {@link #getOutputStream()} should be closed after this method returns.
* <p>
* TODO: Check whether the input and output streams are to be closed as well?
*
* @throws IOException if an I/O error occurs when closing the transport
*/
public func stop(): Unit {
socket?.close()
}
/**
* Returns the URI of the broker which was used to create this NetworkModule.
*
* @return the URI of the broker
*/
public func getServerURI(): String {
return "tcp://${brokerUrl.hostName}:${brokerUrl.port}"
}
}
public class TLSNetworkModule <: NetworkModule {
private static let CLASS_NAME = "TLSNetworkModule"
private let log = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
private let brokerUrl: URL
private let factory: SocketFactory
private let cfg: TlsClientConfig
private let option: MqttConnectOptions
private var socket: ?TcpSocket = None
private var tlsSocket: ?TlsSocket = None
public init(
brokerUrl: URL,
resourceContext: String,
factory!: SocketFactory,
cfg!: TlsClientConfig,
option!: MqttConnectOptions
) {
this.brokerUrl = brokerUrl
this.factory = factory
this.cfg = cfg
this.option = option
log.setResourceName(resourceContext)
}
public func start(): Unit {
this.socket = this.factory(this.getServerURI())
this.socket?.connect(timeout: option.getConnectionTimeout())
this.tlsSocket = TlsSocket.client(socket(), clientConfig: cfg)
this.tlsSocket?.handshake(timeout: option.getConnectionTimeout())
this.tlsSocket?.readTimeout = option.getKeepAliveInterval() * 2
}
public func getInputStream(): InputStream {
this.tlsSocket()
}
public func getOutputStream(): OutputStream {
this.tlsSocket()
}
public func stop(): Unit {
this.tlsSocket?.close()
}
public func getServerURI(): String {
return "ssl://${brokerUrl.hostName}:${brokerUrl.port}"
}
}
public open class WebSocketNetworkModule <: NetworkModule & IOStream {
static let SUB_PROTOCOL = ArrayList<String>(["mqtt"])
private static let CLASS_NAME = "WebSocketNetworkModule"
private let log = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
let brokerUrl: URL
let factory: ?SocketFactory
let option: MqttConnectOptions
let customWebsocketHeaders: ?HashMap<String, String>
let pipedStream: PipedStream
var webSocketReceiver: ?WebSocketReceiver = None
var websocket: ?WebSocket = None
public init(
brokerUrl: URL,
resourceContext: String,
factory!: ?SocketFactory = None,
option!: MqttConnectOptions,
customWebsocketHeaders!: ?HashMap<String, String> = None
) {
this.brokerUrl = brokerUrl
this.factory = factory
this.option = option
this.customWebsocketHeaders = customWebsocketHeaders
this.pipedStream = PipedStream()
log.setResourceName(resourceContext)
}
public open func start(): Unit {
let clientBuilder = ClientBuilder()
clientBuilder.readTimeout(option.getKeepAliveInterval() * 2)
let client = clientBuilder.build()
let httpHeaders = HttpHeaders()
if (let Some(map) <- customWebsocketHeaders) {
for ((k, v) in map) {
httpHeaders.set(k, v)
}
}
let fut: Future<WebSocket> = spawn {
let (futWebsocket, _) = WebSocket.upgradeFromClient(
client,
brokerUrl,
subProtocols: SUB_PROTOCOL,
headers: httpHeaders
)
return futWebsocket
}
let futOp = try {
fut.get(option.getConnectionTimeout())
} catch (_: TimeoutException) {
throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR)
}
this.websocket = futOp
this.webSocketReceiver = WebSocketReceiver(websocket(), pipedStream)
webSocketReceiver?.start("webSocketReceiver")
}
public func getInputStream(): InputStream {
return this
}
public func getOutputStream(): OutputStream {
return this
}
public func read(buf: Array<Byte>): Int64 {
return pipedStream.read(buf)
}
public func write(buf: Array<Byte>): Unit {
websocket?.write(WebSocketFrameType.BinaryWebFrame, buf)
}
public func stop(): Unit {
webSocketReceiver?.stop()
// Creating Close Frame
websocket?.writeCloseFrame(status: 1000)
websocket?.closeConn()
}
public open func getServerURI(): String {
return "ws://${brokerUrl.hostName}:${brokerUrl.port}"
}
}
public class WebSocketSecureNetworkModule <: WebSocketNetworkModule {
private static let CLASS_NAME = "WebSocketSecureNetworkModule"
private let log = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
private let cfg: TlsClientConfig
public init(
brokerUrl: URL,
clientId: String,
factory!: ?SocketFactory = None,
cfg!: TlsClientConfig = DEFAULT_TLS_CLIENT_CONFIG,
option!: MqttConnectOptions,
customWebsocketHeaders!: ?HashMap<String, String> = None
) {
super(
brokerUrl,
clientId,
factory: factory,
customWebsocketHeaders: customWebsocketHeaders,
option: option
)
this.cfg = cfg
log.setResourceName(clientId)
}
public func start(): Unit {
let clientBuilder = ClientBuilder()
clientBuilder.readTimeout(option.getKeepAliveInterval() * 2)
let client = clientBuilder.tlsConfig(cfg).build()
let httpHeaders = HttpHeaders()
if (let Some(map) <- customWebsocketHeaders) {
for ((k, v) in map) {
httpHeaders.set(k, v)
}
}
let fut: Future<WebSocket> = spawn {
let (futWebsocket, _) = WebSocket.upgradeFromClient(
client,
brokerUrl,
subProtocols: SUB_PROTOCOL,
headers: httpHeaders
)
return futWebsocket
}
let futOp = try {
fut.get(option.getKeepAliveInterval() * 2)
} catch (_: TimeoutException) {
throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR)
}
this.websocket = futOp
this.webSocketReceiver = WebSocketReceiver(websocket(), pipedStream)
webSocketReceiver?.start("WssSocketReceiver")
}
public func getServerURI(): String {
return "wss://${brokerUrl.hostName}:${brokerUrl.port}"
}
}