aboutsummaryrefslogtreecommitdiffstats
path: root/facebook/facebook-mqtt.c
diff options
context:
space:
mode:
authorjgeboski <jgeboski@gmail.com>2015-08-24 17:01:16 -0400
committerjgeboski <jgeboski@gmail.com>2015-08-24 17:01:16 -0400
commit53cedfaca78278a64cc347e8ed51bb67363b3dcc (patch)
treedd2c6cac0c3d5d4af1cccaa9cbbc7a6ca4522572 /facebook/facebook-mqtt.c
parent60a42307f24c617029efbb7bcd432d162706c692 (diff)
downloadbitlbee-facebook-53cedfaca78278a64cc347e8ed51bb67363b3dcc.tar.gz
bitlbee-facebook-53cedfaca78278a64cc347e8ed51bb67363b3dcc.tar.bz2
bitlbee-facebook-53cedfaca78278a64cc347e8ed51bb67363b3dcc.tar.xz
Imported changes from purple-facebook
Diffstat (limited to 'facebook/facebook-mqtt.c')
-rw-r--r--facebook/facebook-mqtt.c1259
1 files changed, 590 insertions, 669 deletions
diff --git a/facebook/facebook-mqtt.c b/facebook/facebook-mqtt.c
index 6e85b29..de0fee6 100644
--- a/facebook/facebook-mqtt.c
+++ b/facebook/facebook-mqtt.c
@@ -21,317 +21,405 @@
#include <stdarg.h>
#include <string.h>
+#include "facebook-marshal.h"
#include "facebook-mqtt.h"
+#include "facebook-util.h"
-/**
- * Gets the error domain for #fb_mqtt.
- *
- * @return The #GQuark of the error domain.
- **/
-GQuark fb_mqtt_error_quark(void)
+struct _FbMqttPrivate
{
- static GQuark q;
+ gpointer ssl;
+ gboolean connected;
+ guint16 mid;
- if (G_UNLIKELY(q == 0))
- q = g_quark_from_static_string("fb-mqtt-error-quark");
+ GByteArray *rbuf;
+ GByteArray *wbuf;
+ gsize remz;
- return q;
-}
+ gint tev;
+ gint rev;
+ gint wev;
+};
-/**
- * Creates a new #fb_mqtt. The returned #fb_mqtt should be freed with
- * #fb_mqtt_free() when no longer needed.
- *
- * @param funcs The #fb_mqtt_funcs.
- * @param data The user defined data or NULL.
- *
- * @return The #fb_mqtt or NULL on error.
- **/
-fb_mqtt_t *fb_mqtt_new(const fb_mqtt_funcs_t *funcs, gpointer data)
+struct _FbMqttMessagePrivate
{
- fb_mqtt_t *mqtt;
+ FbMqttMessageType type;
+ FbMqttMessageFlags flags;
- mqtt = g_new0(fb_mqtt_t, 1);
- memcpy(&mqtt->funcs, funcs, sizeof *funcs);
- mqtt->data = data;
- mqtt->rbuf = g_byte_array_new();
- mqtt->wbuf = g_byte_array_new();
+ GByteArray *bytes;
+ guint offset;
+ guint pos;
- return mqtt;
+ gboolean local;
};
-/**
- * Frees all memory used by a #fb_mqtt.
- *
- * @param api The #fb_mqtt.
- **/
-void fb_mqtt_free(fb_mqtt_t *mqtt)
+G_DEFINE_TYPE(FbMqtt, fb_mqtt, G_TYPE_OBJECT);
+G_DEFINE_TYPE(FbMqttMessage, fb_mqtt_message, G_TYPE_OBJECT);
+
+static void
+fb_mqtt_dispose(GObject *obj)
{
- if (G_UNLIKELY(mqtt == NULL))
- return;
+ FbMqtt *mqtt = FB_MQTT(obj);
+ FbMqttPrivate *priv = mqtt->priv;
fb_mqtt_close(mqtt);
- g_clear_error(&mqtt->err);
+ g_byte_array_free(priv->rbuf, TRUE);
+ g_byte_array_free(priv->wbuf, TRUE);
+}
+
+static void
+fb_mqtt_class_init(FbMqttClass *klass)
+{
+ GObjectClass *gklass = G_OBJECT_CLASS(klass);
+
+ gklass->dispose = fb_mqtt_dispose;
+ g_type_class_add_private(klass, sizeof (FbMqttPrivate));
+
+ /**
+ * FbMqtt::connect:
+ * @mqtt: The #FbMqtt.
+ *
+ * Emitted upon the successful completion of the connection
+ * process. This is emitted as a result of #fb_mqtt_connect().
+ */
+ g_signal_new("connect",
+ G_TYPE_FROM_CLASS(klass),
+ G_SIGNAL_ACTION,
+ 0,
+ NULL, NULL,
+ fb_marshal_VOID__VOID,
+ G_TYPE_NONE,
+ 0);
+
+ /**
+ * FbMqtt::error:
+ * @mqtt: The #FbMqtt.
+ * @error: The #GError.
+ *
+ * Emitted whenever an error is hit within the #FbMqtt. This should
+ * close the #FbMqtt with #fb_mqtt_close().
+ */
+ g_signal_new("error",
+ G_TYPE_FROM_CLASS(klass),
+ G_SIGNAL_ACTION,
+ 0,
+ NULL, NULL,
+ fb_marshal_VOID__OBJECT,
+ G_TYPE_NONE,
+ 1, G_TYPE_ERROR);
+
+ /**
+ * FbMqtt::open:
+ * @mqtt: The #FbMqtt.
+ *
+ * Emitted upon the successful opening of the remote socket.
+ * This is emitted as a result of #fb_mqtt_open(). This should
+ * call #fb_mqtt_connect().
+ */
+ g_signal_new("open",
+ G_TYPE_FROM_CLASS(klass),
+ G_SIGNAL_ACTION,
+ 0,
+ NULL, NULL,
+ fb_marshal_VOID__VOID,
+ G_TYPE_NONE,
+ 0);
+
+ /**
+ * FbMqtt::publish:
+ * @mqtt: The #FbMqtt.
+ * @topic: The topic.
+ * @pload: The payload.
+ *
+ * Emitted upon an incoming message from the steam.
+ */
+ g_signal_new("publish",
+ G_TYPE_FROM_CLASS(klass),
+ G_SIGNAL_ACTION,
+ 0,
+ NULL, NULL,
+ fb_marshal_VOID__STRING_BOXED,
+ G_TYPE_NONE,
+ 2, G_TYPE_STRING, G_TYPE_BYTE_ARRAY);
+}
+
+static void
+fb_mqtt_init(FbMqtt *mqtt)
+{
+ FbMqttPrivate *priv;
+
+ priv = G_TYPE_INSTANCE_GET_PRIVATE(mqtt, FB_TYPE_MQTT, FbMqttPrivate);
+ mqtt->priv = priv;
+
+ priv->rbuf = g_byte_array_new();
+ priv->wbuf = g_byte_array_new();
+}
+
+static void
+fb_mqtt_message_dispose(GObject *obj)
+{
+ FbMqttMessagePrivate *priv = FB_MQTT_MESSAGE(obj)->priv;
+
+ if ((priv->bytes != NULL) && priv->local) {
+ g_byte_array_free(priv->bytes, TRUE);
+ }
+}
- g_byte_array_free(mqtt->wbuf, TRUE);
- g_byte_array_free(mqtt->rbuf, TRUE);
+static void
+fb_mqtt_message_class_init(FbMqttMessageClass *klass)
+{
+ GObjectClass *gklass = G_OBJECT_CLASS(klass);
- g_free(mqtt);
+ gklass->dispose = fb_mqtt_message_dispose;
+ g_type_class_add_private(klass, sizeof (FbMqttMessagePrivate));
}
-/**
- * Closes the #fb_mqtt connection.
- *
- * @param mqtt The #fb_mqtt.
- **/
-void fb_mqtt_close(fb_mqtt_t *mqtt)
+static void
+fb_mqtt_message_init(FbMqttMessage *msg)
+{
+ FbMqttMessagePrivate *priv;
+
+ priv = G_TYPE_INSTANCE_GET_PRIVATE(msg, FB_TYPE_MQTT_MESSAGE,
+ FbMqttMessagePrivate);
+ msg->priv = priv;
+}
+
+GQuark
+fb_mqtt_error_quark(void)
{
- g_return_if_fail(mqtt != NULL);
+ static GQuark q = 0;
- if (mqtt->wev > 0) {
- b_event_remove(mqtt->wev);
- mqtt->wev = 0;
+ if (G_UNLIKELY(q == 0)) {
+ q = g_quark_from_static_string("fb-mqtt-error-quark");
}
- if (mqtt->rev > 0) {
- b_event_remove(mqtt->rev);
- mqtt->rev = 0;
+ return q;
+}
+
+GQuark
+fb_mqtt_ssl_error_quark(void)
+{
+ static GQuark q = 0;
+
+ if (G_UNLIKELY(q == 0)) {
+ q = g_quark_from_static_string("fb-mqtt-ssl-error-quark");
}
- if (mqtt->tev > 0) {
- b_event_remove(mqtt->tev);
- mqtt->tev = 0;
+ return q;
+}
+
+FbMqtt *
+fb_mqtt_new(void)
+{
+ return g_object_new(FB_TYPE_MQTT, NULL);
+};
+
+void
+fb_mqtt_close(FbMqtt *mqtt)
+{
+ FbMqttPrivate *priv;
+
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ priv = mqtt->priv;
+
+ if (priv->wev > 0) {
+ b_event_remove(priv->wev);
+ priv->wev = 0;
}
- if (mqtt->ssl != NULL) {
- ssl_disconnect(mqtt->ssl);
- mqtt->ssl = NULL;
+ if (priv->rev > 0) {
+ b_event_remove(priv->rev);
+ priv->rev = 0;
}
-#ifdef DEBUG_FACEBOOK
- if (mqtt->wbuf->len > 0)
- FB_UTIL_DEBUGLN("Closing with unwritten data");
-#endif /* DEBUG_FACEBOOK */
+ if (priv->tev > 0) {
+ b_event_remove(priv->tev);
+ priv->tev = 0;
+ }
+
+ if (priv->ssl != NULL) {
+ ssl_disconnect(priv->ssl);
+ priv->ssl = NULL;
+ }
- mqtt->connected = FALSE;
- g_clear_error(&mqtt->err);
+ if (priv->wbuf->len > 0) {
+ fb_util_debug_warn("Closing with unwritten data");
+ }
- g_byte_array_set_size(mqtt->rbuf, 0);
- g_byte_array_set_size(mqtt->wbuf, 0);
+ priv->connected = FALSE;
+ g_byte_array_set_size(priv->rbuf, 0);
+ g_byte_array_set_size(priv->wbuf, 0);
}
-/**
- * Handles an error with the #fb_mqtt. This sets #fb_mqtt->err, calls
- * the error function, and closes the connection.
- *
- * @param mqtt The #fb_mqtt.
- * @param error The #fb_mqtt_error.
- * @param fmt The format string.
- * @param ... The arguments for the format string.
- **/
-void fb_mqtt_error(fb_mqtt_t *mqtt, fb_mqtt_error_t err, const gchar *fmt, ...)
+void
+fb_mqtt_error(FbMqtt *mqtt, FbMqttError error, const gchar *format, ...)
{
- gchar *str;
- va_list ap;
-
- g_return_if_fail(mqtt != NULL);
+ GError *err;
+ va_list ap;
- if (fmt != NULL) {
- va_start(ap, fmt);
- str = g_strdup_vprintf(fmt, ap);
- va_end(ap);
+ g_return_if_fail(FB_IS_MQTT(mqtt));
- g_clear_error(&mqtt->err);
- g_set_error_literal(&mqtt->err, FB_MQTT_ERROR, err, str);
- g_free(str);
- }
+ va_start(ap, format);
+ err = g_error_new_valist(FB_MQTT_ERROR, error, format, ap);
+ va_end(ap);
- if (mqtt->err != NULL)
- FB_MQTT_FUNC(mqtt, error, mqtt->err);
+ g_signal_emit_by_name(mqtt, "error", err);
+ g_error_free(err);
}
-/**
- * Implemented #b_event_handler for #fb_mqtt_timeout().
- *
- * @param data The user defined data, which is #fb_mqtt.
- * @param fd The event file descriptor.
- * @param cond The #b_input_condition.
- *
- * @return FALSE to prevent continued event handling.
- **/
-static gboolean fb_mqtt_cb_timeout(gpointer data, gint fd,
- b_input_condition cond)
+static gboolean
+fb_mqtt_cb_timeout(gpointer data, gint fd, b_input_condition cond)
{
- fb_mqtt_t *mqtt = data;
+ FbMqtt *mqtt = data;
+ FbMqttPrivate *priv = mqtt->priv;
- mqtt->tev = 0;
+ priv->tev = 0;
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Connection timed out");
return FALSE;
}
-/**
- * Clears an enacted connection timeout.
- *
- * @param mqtt The #fb_mqtt.
- **/
-static void fb_mqtt_timeout_clear(fb_mqtt_t *mqtt)
+static void
+fb_mqtt_timeout_clear(FbMqtt *mqtt)
{
- g_return_if_fail(mqtt != NULL);
+ FbMqttPrivate *priv = mqtt->priv;
- if (mqtt->tev > 0) {
- b_event_remove(mqtt->tev);
- mqtt->tev = 0;
+ if (priv->tev > 0) {
+ b_event_remove(priv->tev);
+ priv->tev = 0;
}
}
-/**
- * Enacts a timeout on the connection. This clears any timeout which
- * currently exists.
- *
- * @param mqtt The #fb_mqtt.
- **/
-static void fb_mqtt_timeout(fb_mqtt_t *mqtt)
+static void
+fb_mqtt_timeout(FbMqtt *mqtt)
{
- g_return_if_fail(mqtt != NULL);
+ FbMqttPrivate *priv = mqtt->priv;
fb_mqtt_timeout_clear(mqtt);
- mqtt->tev = b_timeout_add(FB_MQTT_TIMEOUT_CONN, fb_mqtt_cb_timeout, mqtt);
+ priv->tev = b_timeout_add(FB_MQTT_TIMEOUT_CONN, fb_mqtt_cb_timeout, mqtt);
}
-/**
- * Implemented #b_event_handler for sending a PING request.
- *
- * @param data The user defined data, which is #fb_mqtt.
- * @param fd The event file descriptor.
- * @param cond The #b_input_condition.
- *
- * @return FALSE to prevent continued event handling.
- **/
-static gboolean fb_mqtt_cb_ping(gpointer data, gint fd,
- b_input_condition cond)
+static gboolean
+fb_mqtt_cb_ping(gpointer data, gint fd, b_input_condition cond)
{
- fb_mqtt_t *mqtt = data;
- fb_mqtt_msg_t *msg;
+ FbMqtt *mqtt = data;
+ FbMqttMessage *msg;
+ FbMqttPrivate *priv = mqtt->priv;
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_PINGREQ, 0);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_PINGREQ, 0);
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
- mqtt->tev = 0;
+ priv->tev = 0;
fb_mqtt_timeout(mqtt);
return FALSE;
}
-/**
- * Sends a PING after #FB_MQTT_KA seconds. This clears any timeout which
- * currently exists.
- *
- * @param mqtt The #fb_mqtt.
- **/
-static void fb_mqtt_ping(fb_mqtt_t *mqtt)
+static void
+fb_mqtt_ping(FbMqtt *mqtt)
{
- g_return_if_fail(mqtt != NULL);
+ FbMqttPrivate *priv = mqtt->priv;
fb_mqtt_timeout_clear(mqtt);
- mqtt->tev = b_timeout_add(FB_MQTT_TIMEOUT_PING, fb_mqtt_cb_ping, mqtt);
+ priv->tev = b_timeout_add(FB_MQTT_TIMEOUT_PING, fb_mqtt_cb_ping, mqtt);
}
-/**
- * Implemented #b_event_handler for the read of #fb_mqtt->fd.
- *
- * @param data The user defined data, which is #fb_mqtt.
- * @param fd The event file descriptor.
- * @param cond The #b_input_condition.
- *
- * @return TRUE for continued event handling, otherwise FALSE.
- **/
-static gboolean fb_mqtt_cb_read(gpointer data, gint fd,
- b_input_condition cond)
+static gboolean
+fb_mqtt_cb_read(gpointer data, gint fd, b_input_condition cond)
{
- fb_mqtt_t *mqtt = data;
- fb_mqtt_msg_t *msg;
- gchar buf[1024];
- guint8 byte;
- guint mult;
- gssize rize;
- gint res;
-
- if (mqtt->remz < 1) {
+ FbMqtt *mqtt = data;
+ FbMqttMessage *msg;
+ FbMqttPrivate *priv = mqtt->priv;
+ gint res;
+ guint mult;
+ guint8 buf[1024];
+ guint8 byte;
+ gsize size;
+ gssize rize;
+
+ if (priv->remz < 1) {
/* Reset the read buffer */
- g_byte_array_set_size(mqtt->rbuf, 0);
+ g_byte_array_set_size(priv->rbuf, 0);
- res = ssl_read(mqtt->ssl, (gchar*) &byte, sizeof byte);
- g_byte_array_append(mqtt->rbuf, &byte, sizeof byte);
+ res = ssl_read(priv->ssl, (gchar *) &byte, sizeof byte);
+ g_byte_array_append(priv->rbuf, &byte, sizeof byte);
- if (res != sizeof byte)
- goto error;
+ if (res != sizeof byte) {
+ fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
+ "Failed to read fixed header");
+ return FALSE;
+ }
mult = 1;
do {
- res = ssl_read(mqtt->ssl, (gchar*) &byte, sizeof byte);
- g_byte_array_append(mqtt->rbuf, &byte, sizeof byte);
+ res = ssl_read(priv->ssl, (gchar *) &byte, sizeof byte);
+ g_byte_array_append(priv->rbuf, &byte, sizeof byte);
- if (res != sizeof byte)
- goto error;
+ if (res != sizeof byte) {
+ fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
+ "Failed to read packet size");
+ return FALSE;
+ }
- mqtt->remz += (byte & 127) * mult;
+ priv->remz += (byte & 127) * mult;
mult *= 128;
} while ((byte & 128) != 0);
}
- if (mqtt->remz > 0) {
- rize = ssl_read(mqtt->ssl, buf, MIN(mqtt->remz, sizeof buf));
+ if (priv->remz > 0) {
+ size = MIN(priv->remz, sizeof buf);
+ rize = ssl_read(priv->ssl, (gchar *) buf, size);
- if (rize < 1)
- goto error;
+ if (rize < 1) {
+ fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
+ "Failed to read packet data");
+ return FALSE;
+ }
- g_byte_array_append(mqtt->rbuf, (guint8*) buf, rize);
- mqtt->remz -= rize;
+ g_byte_array_append(priv->rbuf, buf, rize);
+ priv->remz -= rize;
}
- if (mqtt->remz < 1) {
- msg = fb_mqtt_msg_new_bytes(mqtt->rbuf);
- mqtt->remz = 0;
+ if (priv->remz < 1) {
+ msg = fb_mqtt_message_new_bytes(priv->rbuf);
+ priv->remz = 0;
- if (G_UNLIKELY(msg == NULL))
- goto error;
+ if (G_UNLIKELY(msg == NULL)) {
+ fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
+ "Failed to parse message");
+ return FALSE;
+ }
fb_mqtt_read(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
}
return TRUE;
-
-error:
- fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Short read");
- return FALSE;
}
-/**
- * Read a #GByteArray to the #fb_mqtt.
- *
- * @param mqtt The #fb_mqtt.
- * @param bytes The #GByteArray.
- **/
-void fb_mqtt_read(fb_mqtt_t *mqtt, fb_mqtt_msg_t *msg)
+void
+fb_mqtt_read(FbMqtt *mqtt, FbMqttMessage *msg)
{
- fb_mqtt_msg_t *nsg;
- GByteArray *wytes;
- gchar *str;
- guint8 chr;
- guint16 mid;
-
- g_return_if_fail(mqtt != NULL);
- g_return_if_fail(msg != NULL);
-
- fb_util_hexdump(msg->bytes, 2, "Reading %d (flags: 0x%0X)",
- msg->type, msg->flags);
-
- switch (msg->type) {
- case FB_MQTT_MSG_TYPE_CONNACK:
- if (!fb_mqtt_msg_read_byte(msg, NULL) ||
- !fb_mqtt_msg_read_byte(msg, &chr))
+ FbMqttMessage *nsg;
+ FbMqttPrivate *priv;
+ FbMqttMessagePrivate *mriv;
+ GByteArray *wytes;
+ gchar *str;
+ guint8 chr;
+ guint16 mid;
+
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ g_return_if_fail(FB_IS_MQTT_MESSAGE(msg));
+ priv = mqtt->priv;
+ mriv = msg->priv;
+
+ fb_util_debug_hexdump(FB_UTIL_DEBUG_LEVEL_INFO, mriv->bytes,
+ "Reading %d (flags: 0x%0X)",
+ mriv->type, mriv->flags);
+
+ switch (mriv->type) {
+ case FB_MQTT_MESSAGE_TYPE_CONNACK:
+ if (!fb_mqtt_message_read_byte(msg, NULL) ||
+ !fb_mqtt_message_read_byte(msg, &chr))
{
break;
}
@@ -341,62 +429,67 @@ void fb_mqtt_read(fb_mqtt_t *mqtt, fb_mqtt_msg_t *msg)
return;
}
- mqtt->connected = TRUE;
+ priv->connected = TRUE;
fb_mqtt_ping(mqtt);
- FB_MQTT_FUNC(mqtt, connack);
+ g_signal_emit_by_name(mqtt, "connect");
return;
- case FB_MQTT_MSG_TYPE_PUBLISH:
- if (!fb_mqtt_msg_read_str(msg, &str))
+ case FB_MQTT_MESSAGE_TYPE_PUBLISH:
+ if (!fb_mqtt_message_read_str(msg, &str)) {
break;
+ }
- if ((msg->flags & FB_MQTT_MSG_FLAG_QOS1) ||
- (msg->flags & FB_MQTT_MSG_FLAG_QOS2))
+ if ((mriv->flags & FB_MQTT_MESSAGE_FLAG_QOS1) ||
+ (mriv->flags & FB_MQTT_MESSAGE_FLAG_QOS2))
{
- if (msg->flags & FB_MQTT_MSG_FLAG_QOS1)
- chr = FB_MQTT_MSG_TYPE_PUBACK;
- else
- chr = FB_MQTT_MSG_TYPE_PUBREC;
+ if (mriv->flags & FB_MQTT_MESSAGE_FLAG_QOS1) {
+ chr = FB_MQTT_MESSAGE_TYPE_PUBACK;
+ } else {
+ chr = FB_MQTT_MESSAGE_TYPE_PUBREC;
+ }
- if (!fb_mqtt_msg_read_mid(msg, &mid))
+ if (!fb_mqtt_message_read_mid(msg, &mid)) {
break;
+ }
- nsg = fb_mqtt_msg_new(chr, 0);
- fb_mqtt_msg_write_u16(nsg, mid);
+ nsg = fb_mqtt_message_new(chr, 0);
+ fb_mqtt_message_write_u16(nsg, mid);
fb_mqtt_write(mqtt, nsg);
- fb_mqtt_msg_free(nsg);
+ g_object_unref(nsg);
}
wytes = g_byte_array_new();
- fb_mqtt_msg_read_r(msg, wytes);
- FB_MQTT_FUNC(mqtt, publish, str, wytes);
+ fb_mqtt_message_read_r(msg, wytes);
+ g_signal_emit_by_name(mqtt, "publish", str, wytes);
g_byte_array_free(wytes, TRUE);
g_free(str);
return;
- case FB_MQTT_MSG_TYPE_PUBREL:
- if (!fb_mqtt_msg_read_mid(msg, &mid))
+ case FB_MQTT_MESSAGE_TYPE_PUBREL:
+ if (!fb_mqtt_message_read_mid(msg, &mid)) {
break;
+ }
- nsg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_PUBCOMP, 0);
- fb_mqtt_msg_write_u16(nsg, mid); /* Message identifier */
+ nsg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_PUBCOMP, 0);
+ fb_mqtt_message_write_u16(nsg, mid); /* Message identifier */
fb_mqtt_write(mqtt, nsg);
- fb_mqtt_msg_free(nsg);
+ g_object_unref(nsg);
return;
- case FB_MQTT_MSG_TYPE_PINGRESP:
+ case FB_MQTT_MESSAGE_TYPE_PINGRESP:
fb_mqtt_ping(mqtt);
return;
- case FB_MQTT_MSG_TYPE_PUBACK:
- case FB_MQTT_MSG_TYPE_PUBCOMP:
- case FB_MQTT_MSG_TYPE_SUBACK:
- case FB_MQTT_MSG_TYPE_UNSUBACK:
+ case FB_MQTT_MESSAGE_TYPE_PUBACK:
+ case FB_MQTT_MESSAGE_TYPE_PUBCOMP:
+ case FB_MQTT_MESSAGE_TYPE_SUBACK:
+ case FB_MQTT_MESSAGE_TYPE_UNSUBACK:
return;
default:
- fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Unknown packet (%u)",
- msg->type);
+ fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
+ "Unknown packet (%u)",
+ mriv->type);
return;
}
@@ -404,654 +497,482 @@ void fb_mqtt_read(fb_mqtt_t *mqtt, fb_mqtt_msg_t *msg)
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Failed to parse message");
}
-/**
- * Implemented #b_event_handler for the writing of #fb_mqtt->fd.
- *
- * @param data The user defined data, which is #fb_mqtt.
- * @param fd The event file descriptor.
- * @param cond The #b_input_condition.
- *
- * @return TRUE for continued event handling, otherwise FALSE.
- **/
-static gboolean fb_mqtt_cb_write(gpointer data, gint fd,
- b_input_condition cond)
+static gboolean
+fb_mqtt_cb_write(gpointer data, gint fd, b_input_condition cond)
{
- fb_mqtt_t *mqtt = data;
- gssize wize;
+ FbMqtt *mqtt = data;
+ FbMqttPrivate *priv = mqtt->priv;
+ gssize wize;
- wize = ssl_write(mqtt->ssl, (gchar*) mqtt->wbuf->data, mqtt->wbuf->len);
+ wize = ssl_write(priv->ssl, (gchar *) priv->wbuf->data, priv->wbuf->len);
if (wize < 0) {
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Failed to write data");
return FALSE;
}
- if (wize > 0)
- g_byte_array_remove_range(mqtt->wbuf, 0, wize);
+ if (wize > 0) {
+ g_byte_array_remove_range(priv->wbuf, 0, wize);
+ }
- if (mqtt->wbuf->len < 1) {
- mqtt->wev = 0;
+ if (priv->wbuf->len < 1) {
+ priv->wev = 0;
return FALSE;
}
return TRUE;
}
-/**
- * Writes a #fb_mqtt_msg to the #fb_mqtt.
- *
- * @param mqtt The #fb_mqtt.
- * @param msg The #fb_mqtt_msg.
- **/
-void fb_mqtt_write(fb_mqtt_t *mqtt, fb_mqtt_msg_t *msg)
+void
+fb_mqtt_write(FbMqtt *mqtt, FbMqttMessage *msg)
{
const GByteArray *bytes;
+ FbMqttMessagePrivate *mriv;
+ FbMqttPrivate *priv;
gint fd;
- g_return_if_fail(mqtt != NULL);
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ g_return_if_fail(FB_IS_MQTT_MESSAGE(msg));
+ priv = mqtt->priv;
+ mriv = msg->priv;
- bytes = fb_mqtt_msg_bytes(msg);
+ bytes = fb_mqtt_message_bytes(msg);
if (G_UNLIKELY(bytes == NULL)) {
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Failed to format data");
return;
}
- fb_util_hexdump(bytes, 2, "Writing %d (flags: 0x%0X)",
- msg->type, msg->flags);
+ fb_util_debug_hexdump(FB_UTIL_DEBUG_LEVEL_INFO, mriv->bytes,
+ "Writing %d (flags: 0x%0X)",
+ mriv->type, mriv->flags);
- fd = ssl_getfd(mqtt->ssl);
- g_byte_array_append(mqtt->wbuf, bytes->data, bytes->len);
+ fd = ssl_getfd(priv->ssl);
+ g_byte_array_append(priv->wbuf, bytes->data, bytes->len);
+ fb_mqtt_cb_write(mqtt, fd, B_EV_IO_WRITE);
- if ((mqtt->wev < 1) && fb_mqtt_cb_write(mqtt, fd, B_EV_IO_WRITE))
- mqtt->wev = b_input_add(fd, B_EV_IO_WRITE, fb_mqtt_cb_write, mqtt);
+ if (priv->wev > 0) {
+ priv->wev = b_input_add(fd, B_EV_IO_WRITE, fb_mqtt_cb_write, mqtt);
+ }
}
-/**
- * Implemented #ssl_input_function for the connection of #fb_mqtt->ssl.
- *
- * @param data The user defined data, which is #fb_mqtt.
- * @param error The SSL error. (0 on success)
- * @param ssl The SSL source.
- * @param cond The #b_input_condition.
- *
- * @return TRUE for continued event handling, otherwise FALSE.
- **/
-static gboolean fb_mqtt_cb_open(gpointer data, gint error, gpointer ssl,
- b_input_condition cond)
+static gboolean
+fb_mqtt_cb_open(gpointer data, gint error, gpointer ssl,
+ b_input_condition cond)
{
- fb_mqtt_t *mqtt = data;
- gint fd;
+ FbMqtt *mqtt = data;
+ FbMqttPrivate *priv = mqtt->priv;
+ gint fd;
if ((ssl == NULL) || (error != SSL_OK)) {
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Failed to connect");
return FALSE;
}
+ fd = ssl_getfd(priv->ssl);
fb_mqtt_timeout_clear(mqtt);
- fd = ssl_getfd(mqtt->ssl);
- mqtt->rev = b_input_add(fd, B_EV_IO_READ, fb_mqtt_cb_read, mqtt);
-
- FB_MQTT_FUNC(mqtt, open);
+ priv->rev = b_input_add(fd, B_EV_IO_READ, fb_mqtt_cb_read, mqtt);
+ g_signal_emit_by_name(mqtt, "open");
return FALSE;
}
-/**
- * Opens the connection to the MQTT service.
- *
- * @param mqtt The #fb_mqtt.
- **/
-void fb_mqtt_open(fb_mqtt_t *mqtt, const gchar *host, gint port)
+void
+fb_mqtt_open(FbMqtt *mqtt, const gchar *host, gint port)
{
- g_return_if_fail(mqtt != NULL);
+ FbMqttPrivate *priv;
+
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ priv = mqtt->priv;
fb_mqtt_close(mqtt);
- mqtt->ssl = ssl_connect((gchar*) host, port, TRUE, fb_mqtt_cb_open, mqtt);
+ priv->ssl = ssl_connect((gchar *) host, port, TRUE, fb_mqtt_cb_open, mqtt);
- if (mqtt->ssl == NULL) {
- fb_mqtt_cb_open(mqtt, 1, NULL, 0);
+ if (priv->ssl == NULL) {
+ fb_mqtt_cb_open(mqtt, SSL_NOHANDSHAKE, NULL, 0);
return;
}
fb_mqtt_timeout(mqtt);
}
-/**
- * Connects to the MQTT service. This first establishes an SSL based
- * socket. Then it sends the initial connection packet with optional
- * arguments, which correspond to the flags provided. The arguments
- * must be passed in order: client identifier, will topic, will
- * message, username, and password (not required). The arguments must
- * be in a string format.
- *
- * @param mqtt The #fb_mqtt.
- * @param timeout The keep-alive timeout (seconds).
- * @param flags The #fb_mqtt_connect_flags.
- * @param cid The client identifier.
- * @param ... Additional arguments in order, NULL-terminated.
- **/
-void fb_mqtt_connect(fb_mqtt_t *mqtt, guint8 flags, const gchar *cid, ...)
+void
+fb_mqtt_connect(FbMqtt *mqtt, guint8 flags, const GByteArray *pload)
{
- fb_mqtt_msg_t *msg;
- va_list ap;
- const gchar *str;
-
- g_return_if_fail(mqtt != NULL);
+ FbMqttMessage *msg;
- if (G_UNLIKELY(fb_mqtt_connected(mqtt, FALSE)))
- return;
+ g_return_if_fail(!fb_mqtt_connected(mqtt, FALSE));
+ g_return_if_fail(pload != NULL);
/* Facebook always sends a CONNACK, use QoS1 */
flags |= FB_MQTT_CONNECT_FLAG_QOS1;
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_CONNECT, 0);
- fb_mqtt_msg_write_str(msg, FB_MQTT_NAME); /* Protocol name */
- fb_mqtt_msg_write_byte(msg, FB_MQTT_VERS); /* Protocol version */
- fb_mqtt_msg_write_byte(msg, flags); /* Flags */
- fb_mqtt_msg_write_u16(msg, FB_MQTT_KA); /* Keep alive */
- fb_mqtt_msg_write_str(msg, cid); /* Client identifier */
-
- va_start(ap, cid);
-
- while ((str = va_arg(ap, const gchar*)) != NULL)
- fb_mqtt_msg_write_str(msg, str);
-
- va_end(ap);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_CONNECT, 0);
+ fb_mqtt_message_write_str(msg, FB_MQTT_NAME); /* Protocol name */
+ fb_mqtt_message_write_byte(msg, FB_MQTT_LEVEL); /* Protocol level */
+ fb_mqtt_message_write_byte(msg, flags); /* Flags */
+ fb_mqtt_message_write_u16(msg, FB_MQTT_KA); /* Keep alive */
+ fb_mqtt_message_write(msg, pload->data, pload->len);
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+
fb_mqtt_timeout(mqtt);
+ g_object_unref(msg);
}
-/**
- * Checks the #fb_mqtt connection.
- *
- * @param mqtt The #fb_mqtt.
- * @param error TRUE to error upon no connection, FALSE otherwise.
- *
- * @return TRUE if the #fb_mqtt is connected, FALSE otherwise.
- **/
-gboolean fb_mqtt_connected(fb_mqtt_t *mqtt, gboolean error)
+gboolean
+fb_mqtt_connected(FbMqtt *mqtt, gboolean error)
{
+ FbMqttPrivate *priv;
gboolean connected;
- g_return_val_if_fail(mqtt != NULL, FALSE);
+ g_return_val_if_fail(FB_IS_MQTT(mqtt), FALSE);
+ priv = mqtt->priv;
+ connected = (priv->ssl != NULL) && priv->connected;
- connected = (mqtt->ssl != NULL) && mqtt->connected;
-
- if (!connected && error)
+ if (!connected && error) {
fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL, "Not connected");
+ }
return connected;
}
-/**
- * Disconnects from the MQTT service. This cleanly disconnects from the
- * MQTT services, rather than killing the socket stream. This closes
- * the #fb_mqtt via #fb_mqtt_close().
- *
- * @param mqtt The #fb_mqtt.
- **/
-void fb_mqtt_disconnect(fb_mqtt_t *mqtt)
+void
+fb_mqtt_disconnect(FbMqtt *mqtt)
{
- fb_mqtt_msg_t *msg;
+ FbMqttMessage *msg;
- g_return_if_fail(mqtt != NULL);
-
- if (G_UNLIKELY(!fb_mqtt_connected(mqtt, FALSE)))
+ if (G_UNLIKELY(!fb_mqtt_connected(mqtt, FALSE))) {
return;
+ }
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_DISCONNECT, 0);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_DISCONNECT, 0);
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
fb_mqtt_close(mqtt);
}
-/**
- * Publishes a message to MQTT service.
- *
- * @param mqtt The #fb_mqtt.
- * @param topic The message topic.
- * @param pload The #GByteArray payload or NULL.
- **/
-void fb_mqtt_publish(fb_mqtt_t *mqtt, const gchar *topic,
- const GByteArray *pload)
+void
+fb_mqtt_publish(FbMqtt *mqtt, const gchar *topic, const GByteArray *pload)
{
- fb_mqtt_msg_t *msg;
-
- g_return_if_fail(mqtt != NULL);
+ FbMqttMessage *msg;
+ FbMqttPrivate *priv;
- if (G_UNLIKELY(!fb_mqtt_connected(mqtt, TRUE)))
- return;
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ g_return_if_fail(fb_mqtt_connected(mqtt, FALSE));
+ priv = mqtt->priv;
/* Message identifier not required, but for consistency use QoS1 */
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_PUBLISH, FB_MQTT_MSG_FLAG_QOS1);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_PUBLISH,
+ FB_MQTT_MESSAGE_FLAG_QOS1);
- fb_mqtt_msg_write_str(msg, topic); /* Message topic */
- fb_mqtt_msg_write_mid(msg, &mqtt->mid); /* Message identifier */
+ fb_mqtt_message_write_str(msg, topic); /* Message topic */
+ fb_mqtt_message_write_mid(msg, &priv->mid); /* Message identifier */
- if (pload != NULL)
- fb_mqtt_msg_write(msg, pload->data, pload->len);
+ if (pload != NULL) {
+ fb_mqtt_message_write(msg, pload->data, pload->len);
+ }
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
}
-/**
- * Subscribes to one or more topics.
- *
- * @param mqtt The #fb_mqtt.
- * @param topic1 The first topic name.
- * @param qos1 The first QoS value.
- * @param ... Additional topic names and QoS values, NULL-terminated.
- **/
-void fb_mqtt_subscribe(fb_mqtt_t *mqtt, const gchar *topic1, guint16 qos1, ...)
+void
+fb_mqtt_subscribe(FbMqtt *mqtt, const gchar *topic1, guint16 qos1, ...)
{
- fb_mqtt_msg_t *msg;
- va_list ap;
- const gchar *topic;
- guint16 qos;
-
- g_return_if_fail(mqtt != NULL);
+ const gchar *topic;
+ FbMqttMessage *msg;
+ FbMqttPrivate *priv;
+ guint16 qos;
+ va_list ap;
- if (G_UNLIKELY(!fb_mqtt_connected(mqtt, TRUE)))
- return;
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ g_return_if_fail(fb_mqtt_connected(mqtt, FALSE));
+ priv = mqtt->priv;
/* Facebook requires a message identifier, use QoS1 */
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_SUBSCRIBE, FB_MQTT_MSG_FLAG_QOS1);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_SUBSCRIBE,
+ FB_MQTT_MESSAGE_FLAG_QOS1);
- fb_mqtt_msg_write_mid(msg, &mqtt->mid); /* Message identifier */
- fb_mqtt_msg_write_str(msg, topic1); /* First topics */
- fb_mqtt_msg_write_byte(msg, qos1); /* First QoS value */
+ fb_mqtt_message_write_mid(msg, &priv->mid); /* Message identifier */
+ fb_mqtt_message_write_str(msg, topic1); /* First topics */
+ fb_mqtt_message_write_byte(msg, qos1); /* First QoS value */
va_start(ap, qos1);
while ((topic = va_arg(ap, const gchar*)) != NULL) {
qos = va_arg(ap, guint);
- fb_mqtt_msg_write_str(msg, topic); /* Remaining topics */
- fb_mqtt_msg_write_byte(msg, qos); /* Remaining QoS values */
+ fb_mqtt_message_write_str(msg, topic); /* Remaining topics */
+ fb_mqtt_message_write_byte(msg, qos); /* Remaining QoS values */
}
va_end(ap);
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
}
-/**
- * Unsubscribes from one or more topics.
- *
- * @param mqtt The #fb_mqtt.
- * @param topic1 The first topic name.
- * @param ... Additional topic names, NULL-terminated.
- **/
-void fb_mqtt_unsubscribe(fb_mqtt_t *mqtt, const gchar *topic1, ...)
+void
+fb_mqtt_unsubscribe(FbMqtt *mqtt, const gchar *topic1, ...)
{
- fb_mqtt_msg_t *msg;
- va_list ap;
- const gchar *topic;
+ const gchar *topic;
+ FbMqttMessage *msg;
+ FbMqttPrivate *priv;
+ va_list ap;
- g_return_if_fail(mqtt != NULL);
-
- if (G_UNLIKELY(!fb_mqtt_connected(mqtt, TRUE)))
- return;
+ g_return_if_fail(FB_IS_MQTT(mqtt));
+ g_return_if_fail(fb_mqtt_connected(mqtt, FALSE));
+ priv = mqtt->priv;
/* Facebook requires a message identifier, use QoS1 */
- msg = fb_mqtt_msg_new(FB_MQTT_MSG_TYPE_UNSUBSCRIBE, FB_MQTT_MSG_FLAG_QOS1);
+ msg = fb_mqtt_message_new(FB_MQTT_MESSAGE_TYPE_UNSUBSCRIBE,
+ FB_MQTT_MESSAGE_FLAG_QOS1);
- fb_mqtt_msg_write_mid(msg, &mqtt->mid); /* Message identifier */
- fb_mqtt_msg_write_str(msg, topic1); /* First topic */
+ fb_mqtt_message_write_mid(msg, &priv->mid); /* Message identifier */
+ fb_mqtt_message_write_str(msg, topic1); /* First topic */
va_start(ap, topic1);
- while ((topic = va_arg(ap, const gchar*)) != NULL)
- fb_mqtt_msg_write_str(msg, topic); /* Remaining topics */
+ while ((topic = va_arg(ap, const gchar*)) != NULL) {
+ fb_mqtt_message_write_str(msg, topic); /* Remaining topics */
+ }
va_end(ap);
fb_mqtt_write(mqtt, msg);
- fb_mqtt_msg_free(msg);
+ g_object_unref(msg);
}
-/**
- * Creates a new #fb_mqtt_msg. The returned #fb_mqtt_msg should be
- * freed with #fb_mqtt_msg_free() when no longer needed.
- *
- * @param type The #fb_mqtt_msg_type.
- * @param flags The #fb_mqtt_msg_flags.
- *
- * @return The #fb_mqtt_msg or NULL on error.
- **/
-fb_mqtt_msg_t *fb_mqtt_msg_new(fb_mqtt_msg_type_t type,
- fb_mqtt_msg_flags_t flags)
+FbMqttMessage *
+fb_mqtt_message_new(FbMqttMessageType type, FbMqttMessageFlags flags)
{
- fb_mqtt_msg_t *msg;
+ FbMqttMessage *msg;
+ FbMqttMessagePrivate *priv;
+
+ msg = g_object_new(FB_TYPE_MQTT_MESSAGE, NULL);
+ priv = msg->priv;
- msg = g_new0(fb_mqtt_msg_t, 1);
- msg->type = type;
- msg->flags = flags;
- msg->bytes = g_byte_array_new();
- msg->local = TRUE;
+ priv->type = type;
+ priv->flags = flags;
+ priv->bytes = g_byte_array_new();
+ priv->local = TRUE;
return msg;
}
-/**
- * Creates a new #fb_mqtt_msg from a #GByteArray containing a raw data.
- * The returned #fb_mqtt_msg should be freed with #fb_mqtt_msg_free()
- * when no longer needed. The GByteArray passed to this function MUST
- * remain for the lifetime of the #fb_mqtt_msg.
- *
- * @param bytes The #GByteArray.
- *
- * @return The #fb_mqtt_msg or NULL on error.
- **/
-fb_mqtt_msg_t *fb_mqtt_msg_new_bytes(GByteArray *bytes)
+FbMqttMessage *
+fb_mqtt_message_new_bytes(GByteArray *bytes)
{
- fb_mqtt_msg_t *msg;
- guint8 *byte;
+ FbMqttMessage *msg;
+ FbMqttMessagePrivate *priv;
+ guint8 *byte;
- g_return_val_if_fail(bytes != NULL, NULL);
+ g_return_val_if_fail(bytes != NULL, NULL);
g_return_val_if_fail(bytes->len >= 2, NULL);
- msg = g_new0(fb_mqtt_msg_t, 1);
- msg->bytes = bytes;
- msg->local = FALSE;
+ msg = g_object_new(FB_TYPE_MQTT_MESSAGE, NULL);
+ priv = msg->priv;
- if (bytes->len > 1) {
- msg->type = (*bytes->data & 0xF0) >> 4;
- msg->flags = *bytes->data & 0x0F;
+ priv->bytes = bytes;
+ priv->local = FALSE;
+ priv->type = (*bytes->data & 0xF0) >> 4;
+ priv->flags = *bytes->data & 0x0F;
- /* Skip the fixed header */
- for (byte = msg->bytes->data + 1; (*(byte++) & 128) != 0; );
- msg->offset = byte - bytes->data;
- msg->pos = msg->offset;
- }
+ /* Skip the fixed header */
+ for (byte = priv->bytes->data + 1; (*(byte++) & 128) != 0; );
+ priv->offset = byte - bytes->data;
+ priv->pos = priv->offset;
return msg;
}
-/**
- * Frees all memory used by a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- **/
-void fb_mqtt_msg_free(fb_mqtt_msg_t *msg)
+void
+fb_mqtt_message_reset(FbMqttMessage *msg)
{
- g_return_if_fail(msg != NULL);
-
- if (msg->local)
- g_byte_array_free(msg->bytes, TRUE);
-
- g_free(msg);
-}
+ FbMqttMessagePrivate *priv;
-/**
- * Resets a #fb_mqtt_msg. This resets the cursor and removes any sort
- * of fixed header.
- *
- * @param msg The #fb_mqtt_msg.
- **/
-void fb_mqtt_msg_reset(fb_mqtt_msg_t *msg)
-{
- if (G_UNLIKELY(msg == NULL))
- return;
+ g_return_if_fail(FB_IS_MQTT_MESSAGE(msg));
+ priv = msg->priv;
- if (msg->offset > 0) {
- g_byte_array_remove_range(msg->bytes, 0, msg->offset);
- msg->offset = 0;
- msg->pos = 0;
+ if (priv->offset > 0) {
+ g_byte_array_remove_range(priv->bytes, 0, priv->offset);
+ priv->offset = 0;
+ priv->pos = 0;
}
}
-/**
- * Formats the internal #GByteArray of a #fb_mqtt_msg with the required
- * fixed header for sending over the wire. This set the cursor position
- * to the start of the message data.
- *
- * @param msg The #fb_mqtt_msg.
- *
- * @return The internal #GByteArray.
- **/
-const GByteArray *fb_mqtt_msg_bytes(fb_mqtt_msg_t *msg)
+const GByteArray *
+fb_mqtt_message_bytes(FbMqttMessage *msg)
{
- guint8 sbuf[4];
- guint8 byte;
+ FbMqttMessagePrivate *priv;
+ guint i;
+ guint8 byte;
+ guint8 sbuf[4];
guint32 size;
- guint i;
- g_return_val_if_fail(msg != NULL, NULL);
+ g_return_val_if_fail(FB_IS_MQTT_MESSAGE(msg), NULL);
+ priv = msg->priv;
- size = msg->bytes->len - msg->offset;
- i = 0;
+ i = 0;
+ size = priv->bytes->len - priv->offset;
do {
- if (G_UNLIKELY(i >= G_N_ELEMENTS(sbuf)))
+ if (G_UNLIKELY(i >= G_N_ELEMENTS(sbuf))) {
return NULL;
+ }
- byte = size % 128;
+ byte = size % 128;
size /= 128;
- if (size > 0)
+ if (size > 0) {
byte |= 128;
+ }
sbuf[i++] = byte;
} while (size > 0);
- fb_mqtt_msg_reset(msg);
- g_byte_array_prepend(msg->bytes, sbuf, i);
+ fb_mqtt_message_reset(msg);
+ g_byte_array_prepend(priv->bytes, sbuf, i);
- byte = ((msg->type & 0x0F) << 4) | (msg->flags & 0x0F);
- g_byte_array_prepend(msg->bytes, &byte, sizeof byte);
+ byte = ((priv->type & 0x0F) << 4) | (priv->flags & 0x0F);
+ g_byte_array_prepend(priv->bytes, &byte, sizeof byte);
- msg->pos = (i + 1) * (sizeof byte);
- return msg->bytes;
+ priv->pos = (i + 1) * (sizeof byte);
+ return priv->bytes;
}
-/**
- * Reads raw data from a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- * @param data The data buffer or NULL.
- * @param size The size of data to read.
- *
- * @return TRUE if the data was completely read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read(fb_mqtt_msg_t *msg, gpointer data, guint size)
+gboolean
+fb_mqtt_message_read(FbMqttMessage *msg, gpointer data, guint size)
{
- g_return_val_if_fail(msg != NULL, FALSE);
+ FbMqttMessagePrivate *priv;
+
+ g_return_val_if_fail(FB_IS_MQTT_MESSAGE(msg), FALSE);
+ priv = msg->priv;
- if ((msg->pos + size) > msg->bytes->len)
+ if ((priv->pos + size) > priv->bytes->len) {
return FALSE;
+ }
- if ((data != NULL) && (size > 0))
- memcpy(data, msg->bytes->data + msg->pos, size);
+ if ((data != NULL) && (size > 0)) {
+ memcpy(data, priv->bytes->data + priv->pos, size);
+ }
- msg->pos += size;
+ priv->pos += size;
return TRUE;
}
-/**
- * Reads the remaining bytes from a #fb_mqtt_msg into a #GByteArray.
- *
- * @param msg The #fb_mqtt_msg.
- * @param bytes The #GByteArray.
- *
- * @return TRUE if the byte string was read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read_r(fb_mqtt_msg_t *msg, GByteArray *bytes)
+gboolean
+fb_mqtt_message_read_r(FbMqttMessage *msg, GByteArray *bytes)
{
+ FbMqttMessagePrivate *priv;
guint size;
- g_return_val_if_fail(bytes != NULL, FALSE);
-
- size = msg->bytes->len - msg->pos;
+ g_return_val_if_fail(FB_IS_MQTT_MESSAGE(msg), FALSE);
+ priv = msg->priv;
+ size = priv->bytes->len - priv->pos;
- if (G_LIKELY(size > 0))
- g_byte_array_append(bytes, msg->bytes->data + msg->pos, size);
+ if (G_LIKELY(size > 0)) {
+ g_byte_array_append(bytes, priv->bytes->data + priv->pos,
+ size);
+ }
return TRUE;
}
-/**
- * Reads a single byte from a #fb_mqtt_msg. If the return location is
- * NULL, only the cursor is advanced.
- *
- * @param msg The #fb_mqtt_msg.
- * @param byte The return location for the byte or NULL.
- *
- * @return TRUE if the byte string was read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read_byte(fb_mqtt_msg_t *msg, guint8 *byte)
+gboolean
+fb_mqtt_message_read_byte(FbMqttMessage *msg, guint8 *value)
{
- if (byte != NULL)
- *byte = 0;
-
- return fb_mqtt_msg_read(msg, byte, sizeof *byte);
+ return fb_mqtt_message_read(msg, value, sizeof *value);
}
-/**
- * Reads a message identifier from a #fb_mqtt_msg. If the return
- * location is NULL, only the cursor is advanced.
- *
- * @param msg The #fb_mqtt_msg.
- * @param mid The return location for the message identifier or NULL.
- *
- * @return TRUE if the message identifier was read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read_mid(fb_mqtt_msg_t *msg, guint16 *mid)
+gboolean
+fb_mqtt_message_read_mid(FbMqttMessage *msg, guint16 *value)
{
- return fb_mqtt_msg_read_u16(msg, mid);
+ return fb_mqtt_message_read_u16(msg, value);
}
-/**
- * Reads an unsigned 16-bit integer from a #fb_mqtt_msg. If the return
- * location is NULL, only the cursor is advanced.
- *
- * @param msg The #fb_mqtt_msg.
- * @param u16 The return location for the integer or NULL.
- *
- * @return TRUE if the integer was read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read_u16(fb_mqtt_msg_t *msg, guint16 *u16)
+gboolean
+fb_mqtt_message_read_u16(FbMqttMessage *msg, guint16 *value)
{
- if (!fb_mqtt_msg_read(msg, u16, sizeof *u16)) {
- if (u16 != NULL)
- *u16 = 0;
-
+ if (!fb_mqtt_message_read(msg, value, sizeof *value)) {
return FALSE;
}
- if (u16 != NULL)
- *u16 = g_ntohs(*u16);
+ if (value != NULL) {
+ *value = g_ntohs(*value);
+ }
return TRUE;
}
-/**
- * Reads a string from a #fb_mqtt_msg. If the return location is NULL,
- * only the cursor is advanced. The returned string should be freed
- * with #g_free() when no longer needed.
- *
- * @param msg The #fb_mqtt_msg.
- * @param str The return location for the string or NULL.
- *
- * @return TRUE if the string was read, otherwise FALSE.
- **/
-gboolean fb_mqtt_msg_read_str(fb_mqtt_msg_t *msg, gchar **str)
+gboolean
+fb_mqtt_message_read_str(FbMqttMessage *msg, gchar **value)
{
- guint16 size;
- guint8 *data;
-
- if (str != NULL)
- *str = NULL;
+ guint8 *data;
+ guint16 size;
- if (!fb_mqtt_msg_read_u16(msg, &size))
+ if (!fb_mqtt_message_read_u16(msg, &size)) {
return FALSE;
+ }
- if (str != NULL) {
+ if (value != NULL) {
data = g_new(guint8, size + 1);
data[size] = 0;
} else {
data = NULL;
}
- if (!fb_mqtt_msg_read(msg, data, size)) {
+ if (!fb_mqtt_message_read(msg, data, size)) {
g_free(data);
return FALSE;
}
- if (str != NULL)
- *str = (gchar*) data;
+ if (value != NULL) {
+ *value = (gchar*) data;
+ }
return TRUE;
}
-/**
- * Writes raw data to a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- * @param data The data.
- * @param size The size of the data.
- **/
-void fb_mqtt_msg_write(fb_mqtt_msg_t *msg, gconstpointer data, guint size)
+void
+fb_mqtt_message_write(FbMqttMessage *msg, gconstpointer data, guint size)
{
- g_return_if_fail(msg != NULL);
+ FbMqttMessagePrivate *priv;
- g_byte_array_append(msg->bytes, data, size);
- msg->pos += size;
+ g_return_if_fail(FB_IS_MQTT_MESSAGE(msg));
+ priv = msg->priv;
+
+ g_byte_array_append(priv->bytes, data, size);
+ priv->pos += size;
}
-/**
- * Writes a single byte to a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- * @param byte The byte.
- **/
-void fb_mqtt_msg_write_byte(fb_mqtt_msg_t *msg, guint8 byte)
+void
+fb_mqtt_message_write_byte(FbMqttMessage *msg, guint8 value)
{
- fb_mqtt_msg_write(msg, &byte, sizeof byte);
+ fb_mqtt_message_write(msg, &value, sizeof value);
}
-/**
- * Writes a 16-bit message identifier to a #fb_mqtt_msg. This advances
- * the message identifier by one before usage.
- *
- * @param msg The #fb_mqtt_msg.
- * @param mid The return location of the message identifier.
- **/
-void fb_mqtt_msg_write_mid(fb_mqtt_msg_t *msg, guint16 *mid)
+void
+fb_mqtt_message_write_mid(FbMqttMessage *msg, guint16 *value)
{
- g_return_if_fail(mid != NULL);
-
- fb_mqtt_msg_write_u16(msg, ++(*mid));
+ g_return_if_fail(value != NULL);
+ fb_mqtt_message_write_u16(msg, ++(*value));
}
-/**
- * Writes an unsigned 16-bit integer to a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- * @param u16 Theinteger.
- **/
-void fb_mqtt_msg_write_u16(fb_mqtt_msg_t *msg, guint16 u16)
+void
+fb_mqtt_message_write_u16(FbMqttMessage *msg, guint16 value)
{
- u16 = g_htons(u16);
- fb_mqtt_msg_write(msg, &u16, sizeof u16);
+ value = g_htons(value);
+ fb_mqtt_message_write(msg, &value, sizeof value);
}
-/**
- * Writes a string to a #fb_mqtt_msg.
- *
- * @param msg The #fb_mqtt_msg.
- * @param str The string.
- **/
-void fb_mqtt_msg_write_str(fb_mqtt_msg_t *msg, const gchar *str)
+void
+fb_mqtt_message_write_str(FbMqttMessage *msg, const gchar *value)
{
gint16 size;
- g_return_if_fail(str != NULL);
+ g_return_if_fail(value != NULL);
- size = strlen(str);
- fb_mqtt_msg_write_u16(msg, size);
- fb_mqtt_msg_write(msg, str, size);
+ size = strlen(value);
+ fb_mqtt_message_write_u16(msg, size);
+ fb_mqtt_message_write(msg, value, size);
}