diff --git a/src/mqtt.c b/src/mqtt.c index 08c8569..c6c538e 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -407,9 +407,9 @@ int unpack_mqtt_packet(const unsigned char *raw, union mqtt_packet *pkt) { .byte = type }; - if (header.bits.type == DISCONNECT_TYPE - || header.bits.type == PINGREQ_TYPE - || header.bits.type == PINGRESP_TYPE) + if (header.bits.type == DISCONNECT + || header.bits.type == PINGREQ + || header.bits.type == PINGRESP) pkt->header = header; else /* Call the appropriate unpack handler based on the message type */ @@ -533,7 +533,7 @@ static unsigned char *pack_mqtt_publish(const union mqtt_packet *pkt) { unsigned char *pack_mqtt_packet(const union mqtt_packet *pkt, unsigned type) { - if (type == PINGREQ_TYPE || type == PINGRESP_TYPE) + if (type == PINGREQ || type == PINGRESP) return pack_mqtt_header(&pkt->header); return pack_handlers[type](pkt); @@ -619,7 +619,7 @@ struct mqtt_publish *mqtt_packet_publish(unsigned char byte, void mqtt_packet_release(union mqtt_packet *pkt, unsigned type) { switch (type) { - case CONNECT_TYPE: + case CONNECT: sol_free(pkt->connect.payload.client_id); if (pkt->connect.bits.username == 1) sol_free(pkt->connect.payload.username); @@ -630,16 +630,16 @@ void mqtt_packet_release(union mqtt_packet *pkt, unsigned type) { sol_free(pkt->connect.payload.will_topic); } break; - case SUBSCRIBE_TYPE: - case UNSUBSCRIBE_TYPE: + case SUBSCRIBE: + case UNSUBSCRIBE: for (unsigned i = 0; i < pkt->subscribe.tuples_len; i++) sol_free(pkt->subscribe.tuples[i].topic); sol_free(pkt->subscribe.tuples); break; - case SUBACK_TYPE: + case SUBACK: sol_free(pkt->suback.rcs); break; - case PUBLISH_TYPE: + case PUBLISH: sol_free(pkt->publish.topic); sol_free(pkt->publish.payload); break; diff --git a/src/mqtt.h b/src/mqtt.h index 64687e2..058cba3 100644 --- a/src/mqtt.h +++ b/src/mqtt.h @@ -34,41 +34,36 @@ #define MQTT_HEADER_LEN 2 #define MQTT_ACK_LEN 4 +/* + * Stub bytes, useful for generic replies, these represent the first byte in + * the fixed header + */ +#define CONNACK_BYTE 0x20 +#define PUBLISH_BYTE 0x30 +#define PUBACK_BYTE 0x40 +#define PUBREC_BYTE 0x50 +#define PUBREL_BYTE 0x60 +#define PUBCOMP_BYTE 0x70 +#define SUBACK_BYTE 0x90 +#define UNSUBACK_BYTE 0xB0 +#define PINGRESP_BYTE 0xD0 /* Message types */ -enum message_opcode { - CONNECT = 0x10, - CONNACK = 0x20, - PUBLISH = 0x30, - PUBACK = 0x40, - PUBREC = 0x50, - PUBREL = 0x60, - PUBCOMP = 0x70, - SUBSCRIBE = 0x80, - SUBACK = 0x90, - UNSUBSCRIBE = 0xA0, - UNSUBACK = 0xB0, - PINGREQ = 0xC0, - PINGRESP = 0xD0, - DISCONNECT = 0xE0 -}; - - -enum message_type { - CONNECT_TYPE = 1, - CONNACK_TYPE = 2, - PUBLISH_TYPE = 3, - PUBACK_TYPE = 4, - PUBREC_TYPE = 5, - PUBREL_TYPE = 6, - PUBCOMP_TYPE = 7, - SUBSCRIBE_TYPE = 8, - SUBACK_TYPE = 9, - UNSUBSCRIBE_TYPE = 10, - UNSUBACK_TYPE = 11, - PINGREQ_TYPE = 12, - PINGRESP_TYPE = 13, - DISCONNECT_TYPE = 14 +enum packet_type { + CONNECT = 1, + CONNACK = 2, + PUBLISH = 3, + PUBACK = 4, + PUBREC = 5, + PUBREL = 6, + PUBCOMP = 7, + SUBSCRIBE = 8, + SUBACK = 9, + UNSUBSCRIBE = 10, + UNSUBACK = 11, + PINGREQ = 12, + PINGRESP = 13, + DISCONNECT = 14 }; diff --git a/src/server.c b/src/server.c index eff4357..be7efd0 100644 --- a/src/server.c +++ b/src/server.c @@ -164,7 +164,7 @@ static ssize_t recv_packet(int clientfd, unsigned char *buf, char *command) { unsigned char byte = *buf; buf++; - if (DISCONNECT < byte || CONNECT > byte) + if (DISCONNECT < (byte >> 4) || CONNECT > (byte >> 4)) return -ERRPACKETERR; /* @@ -264,7 +264,7 @@ static int connect_handler(struct closure *cb, union mqtt_packet *pkt) { /* Respond with a connack */ union mqtt_packet *response = sol_malloc(sizeof(*response)); - unsigned char byte = CONNACK; + unsigned char byte = CONNACK_BYTE; // TODO check for session already present @@ -278,7 +278,7 @@ static int connect_handler(struct closure *cb, union mqtt_packet *pkt) { response->connack = *mqtt_packet_connack(byte, connect_flags, rc); cb->payload = bytestring_create(MQTT_ACK_LEN); - unsigned char *p = pack_mqtt_packet(response, CONNACK_TYPE); + unsigned char *p = pack_mqtt_packet(response, CONNACK); memcpy(cb->payload->data, p, MQTT_ACK_LEN); sol_free(p); @@ -390,20 +390,20 @@ static int subscribe_handler(struct closure *cb, union mqtt_packet *pkt) { rcs[i] = pkt->subscribe.tuples[i].qos; } - struct mqtt_suback *suback = mqtt_packet_suback(SUBACK, + struct mqtt_suback *suback = mqtt_packet_suback(SUBACK_BYTE, pkt->subscribe.pkt_id, rcs, pkt->subscribe.tuples_len); - mqtt_packet_release(pkt, SUBSCRIBE_TYPE); + mqtt_packet_release(pkt, SUBSCRIBE); pkt->suback = *suback; - unsigned char *packed = pack_mqtt_packet(pkt, SUBACK_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, SUBACK); size_t len = MQTT_HEADER_LEN + sizeof(uint16_t) + pkt->subscribe.tuples_len; cb->payload = bytestring_create(len); memcpy(cb->payload->data, packed, len); sol_free(packed); - mqtt_packet_release(pkt, SUBACK_TYPE); + mqtt_packet_release(pkt, SUBACK); sol_free(suback); sol_debug("Sending SUBACK to %s", c->client_id); @@ -418,9 +418,9 @@ static int unsubscribe_handler(struct closure *cb, union mqtt_packet *pkt) { sol_debug("Received UNSUBSCRIBE from %s", c->client_id); - pkt->ack = *mqtt_packet_ack(UNSUBACK, pkt->unsubscribe.pkt_id); + pkt->ack = *mqtt_packet_ack(UNSUBACK_BYTE, pkt->unsubscribe.pkt_id); - unsigned char *packed = pack_mqtt_packet(pkt, UNSUBACK_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, UNSUBACK); cb->payload = bytestring_create(MQTT_ACK_LEN); memcpy(cb->payload->data, packed, MQTT_ACK_LEN); sol_free(packed); @@ -492,7 +492,7 @@ static int publish_handler(struct closure *cb, union mqtt_packet *pkt) { if (pkt->publish.header.bits.qos > AT_MOST_ONCE) publen += sizeof(uint16_t); - pub = pack_mqtt_packet(pkt, PUBLISH_TYPE); + pub = pack_mqtt_packet(pkt, PUBLISH); ssize_t sent; if ((sent = send_bytes(sc->fd, pub, publen)) < 0) @@ -520,13 +520,14 @@ static int publish_handler(struct closure *cb, union mqtt_packet *pkt) { if (qos == AT_LEAST_ONCE) { - mqtt_puback *puback = mqtt_packet_ack(PUBACK, pkt->publish.pkt_id); + mqtt_puback *puback = mqtt_packet_ack(PUBACK_BYTE, + pkt->publish.pkt_id); - mqtt_packet_release(pkt, PUBLISH_TYPE); + mqtt_packet_release(pkt, PUBLISH); pkt->ack = *puback; - unsigned char *packed = pack_mqtt_packet(pkt, PUBACK_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, PUBACK); cb->payload = bytestring_create(MQTT_ACK_LEN); memcpy(cb->payload->data, packed, MQTT_ACK_LEN); sol_free(packed); @@ -539,13 +540,13 @@ static int publish_handler(struct closure *cb, union mqtt_packet *pkt) { // TODO add to a hashtable to track PUBREC clients last - mqtt_pubrec *pubrec = mqtt_packet_ack(PUBREC, pkt->publish.pkt_id); + mqtt_pubrec *pubrec = mqtt_packet_ack(PUBREC_BYTE, pkt->publish.pkt_id); - mqtt_packet_release(pkt, PUBLISH_TYPE); + mqtt_packet_release(pkt, PUBLISH); pkt->ack = *pubrec; - unsigned char *packed = pack_mqtt_packet(pkt, PUBREC_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, PUBREC); cb->payload = bytestring_create(MQTT_ACK_LEN); memcpy(cb->payload->data, packed, MQTT_ACK_LEN); sol_free(packed); @@ -556,7 +557,7 @@ static int publish_handler(struct closure *cb, union mqtt_packet *pkt) { } - mqtt_packet_release(pkt, PUBLISH_TYPE); + mqtt_packet_release(pkt, PUBLISH); /* * We're in the case of AT_MOST_ONCE QoS level, we don't need to sent out @@ -583,11 +584,11 @@ static int pubrec_handler(struct closure *cb, union mqtt_packet *pkt) { sol_debug("Received PUBREC from %s", c->client_id); - mqtt_pubrel *pubrel = mqtt_packet_ack(PUBREL, pkt->publish.pkt_id); + mqtt_pubrel *pubrel = mqtt_packet_ack(PUBREL_BYTE, pkt->publish.pkt_id); pkt->ack = *pubrel; - unsigned char *packed = pack_mqtt_packet(pkt, PUBREC_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, PUBREC); cb->payload = bytestring_create(MQTT_ACK_LEN); memcpy(cb->payload->data, packed, MQTT_ACK_LEN); sol_free(packed); @@ -603,11 +604,11 @@ static int pubrel_handler(struct closure *cb, union mqtt_packet *pkt) { sol_debug("Received PUBREL from %s", ((struct sol_client *) cb->obj)->client_id); - mqtt_pubcomp *pubcomp = mqtt_packet_ack(PUBCOMP, pkt->publish.pkt_id); + mqtt_pubcomp *pubcomp = mqtt_packet_ack(PUBCOMP_BYTE, pkt->publish.pkt_id); pkt->ack = *pubcomp; - unsigned char *packed = pack_mqtt_packet(pkt, PUBCOMP_TYPE); + unsigned char *packed = pack_mqtt_packet(pkt, PUBCOMP); cb->payload = bytestring_create(MQTT_ACK_LEN); memcpy(cb->payload->data, packed, MQTT_ACK_LEN); sol_free(packed); @@ -635,8 +636,8 @@ static int pingreq_handler(struct closure *cb, union mqtt_packet *pkt) { sol_debug("Received PINGREQ from %s", ((struct sol_client *) cb->obj)->client_id); - pkt->header = *mqtt_packet_header(PINGRESP); - unsigned char *packed = pack_mqtt_packet(pkt, PINGRESP_TYPE); + pkt->header = *mqtt_packet_header(PINGRESP_BYTE); + unsigned char *packed = pack_mqtt_packet(pkt, PINGRESP); cb->payload = bytestring_create(MQTT_HEADER_LEN); memcpy(cb->payload->data, packed, MQTT_HEADER_LEN); sol_free(packed); @@ -884,7 +885,8 @@ static void publish_message(unsigned short pkt_id, /* Build MQTT packet with command PUBLISH */ union mqtt_packet pkt; - struct mqtt_publish *p = mqtt_packet_publish(PUBLISH, pkt_id, + struct mqtt_publish *p = mqtt_packet_publish(PUBLISH_BYTE, + pkt_id, topiclen, (unsigned char *) topic, payloadlen, @@ -920,7 +922,7 @@ static void publish_message(unsigned short pkt_id, if (pkt.publish.header.bits.qos > AT_MOST_ONCE) len += sizeof(uint16_t); - packed = pack_mqtt_packet(&pkt, PUBLISH_TYPE); + packed = pack_mqtt_packet(&pkt, PUBLISH); if ((sent = send_bytes(sc->fd, packed, len)) < 0) sol_error("Error publishing to %s: %s",