MQTT v5 bridges can handle "retain-available" being false.

eclipse · Nov 7, 2019 · d003fed · d003fed
1 parent 16dc545
commit d003fed
Show file tree

Hide file tree

Showing 4 changed files with 54 additions and 31 deletions.
diff --git a/ChangeLog.txt b/ChangeLog.txt
@@ -6,6 +6,8 @@ Broker:
 - Add `bridge_outgoing_retain` option, to allow outgoing messages from a
  bridge to have the retain bit completely disabled, which is useful when
  bridging to e.g. Amazon or Google.
+- Add support for MQTT v5 bridges to handle the "retain-available" property
+ being false.
 
 Client library:
 - Client no longer generates random client ids for v3.1.1 clients, these are

diff --git a/lib/send_connect.c b/lib/send_connect.c
@@ -149,7 +149,10 @@ int send__connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session
  packet__write_byte(packet, version);
  byte = (clean_session&0x1)<<1;
  if(will){
- byte = byte | ((mosq->will->msg.retain&0x1)<<5) | ((mosq->will->msg.qos&0x3)<<3) | ((will&0x1)<<2);
+ byte = byte | ((mosq->will->msg.qos&0x3)<<3) | ((will&0x1)<<2);
+ if(mosq->retain_available){
+ byte |= (mosq->will->msg.retain&0x1)<<5;
+ }
  }
  if(username){
  byte = byte | 0x1<<7;

diff --git a/src/bridge.c b/src/bridge.c
@@ -447,13 +447,17 @@ int bridge__on_connect(struct mosquitto_db *db, struct mosquitto *context)
  int notification_topic_len;
  char notification_payload;
  int sub_opts;
+ bool retain = true;
 
  if(context->bridge->notifications){
+ if(!context->retain_available){
+ retain = false;
+ }
  notification_payload = '1';
  if(context->bridge->notification_topic){
  if(!context->bridge->notifications_local_only){
  if(send__real_publish(context, mosquitto__mid_generate(context),
- context->bridge->notification_topic, 1, &notification_payload, 1, true, 0, NULL, NULL, 0)){
+ context->bridge->notification_topic, 1, &notification_payload, 1, retain, 0, NULL, NULL, 0)){
 
  return 1;
  }
@@ -468,7 +472,7 @@ int bridge__on_connect(struct mosquitto_db *db, struct mosquitto *context)
  notification_payload = '1';
  if(!context->bridge->notifications_local_only){
  if(send__real_publish(context, mosquitto__mid_generate(context),
- notification_topic, 1, &notification_payload, 1, true, 0, NULL, NULL, 0)){
+ notification_topic, 1, &notification_payload, 1, retain, 0, NULL, NULL, 0)){
 
  mosquitto__free(notification_topic);
  return 1;

diff --git a/src/handle_connack.c b/src/handle_connack.c
@@ -47,37 +47,51 @@ int handle__connack(struct mosquitto_db *db, struct mosquitto *context)
  }
  mosquitto_property_free_all(&properties); /* FIXME - TEMPORARY UNTIL PROPERTIES PROCESSED */
 
- switch(reason_code){
- case CONNACK_ACCEPTED:
+ if(reason_code == MQTT_RC_SUCCESS){
 #ifdef WITH_BRIDGE
-  if(context->bridge){
-  rc = bridge__on_connect(db, context);
-  if(rc) return rc;
-  }
+ if(context->bridge){
+ rc = bridge__on_connect(db, context);
+ if(rc) return rc;
+ }
 #endif
- mosquitto__set_state(context, mosq_cs_active);
- return MOSQ_ERR_SUCCESS;
- case CONNACK_REFUSED_PROTOCOL_VERSION:
- if(context->bridge){
- context->bridge->try_private_accepted = false;
+ mosquitto__set_state(context, mosq_cs_active);
+ return MOSQ_ERR_SUCCESS;
+ }else{
+ if(context->protocol == mosq_p_mqtt5){
+ switch(reason_code){
+ case MQTT_RC_RETAIN_NOT_SUPPORTED:
+ context->retain_available = 0;
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: retain not available (will retry)");
+ return 1;
+ default:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: %s", "FIXME"); //mosquitto_reason_string(reason_code));
+ return 1;
+ }
+ }else{
+ switch(reason_code){
+ case CONNACK_REFUSED_PROTOCOL_VERSION:
+ if(context->bridge){
+ context->bridge->try_private_accepted = false;
+ }
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: unacceptable protocol version");
+ return 1;
+ case CONNACK_REFUSED_IDENTIFIER_REJECTED:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: identifier rejected");
+ return 1;
+ case CONNACK_REFUSED_SERVER_UNAVAILABLE:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
+ return 1;
+ case CONNACK_REFUSED_BAD_USERNAME_PASSWORD:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
+ return 1;
+ case CONNACK_REFUSED_NOT_AUTHORIZED:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: not authorised");
+ return 1;
+ default:
+ log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: unknown reason");
+ return 1;
  }
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: unacceptable protocol version");
- return 1;
- case CONNACK_REFUSED_IDENTIFIER_REJECTED:
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: identifier rejected");
- return 1;
- case CONNACK_REFUSED_SERVER_UNAVAILABLE:
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
- return 1;
- case CONNACK_REFUSED_BAD_USERNAME_PASSWORD:
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
- return 1;
- case CONNACK_REFUSED_NOT_AUTHORIZED:
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: not authorised");
- return 1;
- default:
- log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: unknown reason");
- return 1;
+ }
  }
  return 1;
 }