Coverage for src/updates2mqtt/mqtt.py: 76%

1import asyncio

2import json

3import re

4import time

5from collections.abc import Callable

6from dataclasses import dataclass, field

7from threading import Event

8from typing import Any

10import paho.mqtt.client as mqtt

11import paho.mqtt.subscribeoptions

12import structlog

13from paho.mqtt.client import MQTT_CLEAN_START_FIRST_ONLY, MQTTMessage, MQTTMessageInfo

14from paho.mqtt.enums import CallbackAPIVersion, MQTTErrorCode, MQTTProtocolVersion

15from paho.mqtt.properties import Properties

16from paho.mqtt.reasoncodes import ReasonCode

18from updates2mqtt.model import Discovery, ReleaseProvider

20from .config import HomeAssistantConfig, MqttConfig, NodeConfig, PublishPolicy

21from .hass_formatter import hass_format_config, hass_format_state

23log = structlog.get_logger()

25MQTT_NAME = r"[A-Za-z0-9_\-\.]+"

28@dataclass

29class LocalMessage:

30 topic: str | None = field(default=None)

31 payload: str | None = field(default=None)

34class MqttPublisher:

35 def __init__(self, cfg: MqttConfig, node_cfg: NodeConfig, hass_cfg: HomeAssistantConfig) -> None:

36 self.cfg: MqttConfig = cfg

37 self.node_cfg: NodeConfig = node_cfg

38 self.hass_cfg: HomeAssistantConfig = hass_cfg

39 self.providers_by_topic: dict[str, ReleaseProvider] = {}

40 self.providers_by_type: dict[str, ReleaseProvider] = {}

41 self.event_loop: asyncio.AbstractEventLoop | None = None

42 self.client: mqtt.Client | None = None

43 self.fatal_failure = Event()

44 self.connected = Event()

45 self.commands_in_progress: set[tuple[str, str]] = set()

46 self.log = structlog.get_logger().bind(host=cfg.host, integration="mqtt")

48 def start(self, event_loop: asyncio.AbstractEventLoop | None = None) -> None:

49 logger = self.log.bind(action="start")

50 try:

51 protocol: MQTTProtocolVersion

52 if self.cfg.protocol in ("3", "3.11"):

53 protocol = MQTTProtocolVersion.MQTTv311

54 elif self.cfg.protocol == "3.1":

55 protocol = MQTTProtocolVersion.MQTTv31

56 elif self.cfg.protocol in ("5", "5.0"):

57 protocol = MQTTProtocolVersion.MQTTv5

58 else:

59 logger.info("No valid MQTT protocol version found (%s), setting to default v3.11", self.cfg.protocol)

60 protocol = MQTTProtocolVersion.MQTTv311

61 logger.debug("MQTT protocol set to %r", protocol)

63 self.event_loop = event_loop or asyncio.get_event_loop()

64 self.client = mqtt.Client(

65 callback_api_version=CallbackAPIVersion.VERSION2,

66 client_id=f"updates2mqtt_{self.node_cfg.name}",

67 clean_session=True if protocol != MQTTProtocolVersion.MQTTv5 else None,

68 protocol=protocol,

69 )

70 self.client.username_pw_set(self.cfg.user, password=self.cfg.password)

72 if self.cfg.ca_certs:

73 logger.debug("Configuring TLS, ca_certs=%s,cert_reqs=%s", self.cfg.ca_certs, self.cfg.cert_reqs)

74 self.client.tls_set(

75 ca_certs=self.cfg.ca_certs or None,

76 certfile=self.cfg.client_cert or None,

77 keyfile=self.cfg.client_key or None,

78 cert_reqs=self.cfg.cert_reqs or None,

79 )

80 rc: MQTTErrorCode = self.client.connect(

81 host=self.cfg.host,

82 port=self.cfg.port,

83 keepalive=self.cfg.keepalive,

84 clean_start=MQTT_CLEAN_START_FIRST_ONLY,

85 )

86 logger.info("Client connection requested", result_code=rc)

88 self.client.on_connect = self.on_connect

89 self.client.on_disconnect = self.on_disconnect

90 self.client.on_message = self.on_message

91 self.client.on_subscribe = self.on_subscribe

92 self.client.on_unsubscribe = self.on_unsubscribe

94 self.client.loop_start()

96 if not self.connected.wait(timeout=self.cfg.connect_timeout) and not self.fatal_failure.is_set(): 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true

97 logger.warning("Timed out waiting for broker connection, continuing anyway", timeout=self.cfg.connect_timeout)

99 logger.debug("MQTT Publisher loop started", host=self.cfg.host, port=self.cfg.port)

100 except Exception as e:

101 logger.error("Failed to connect to broker", host=self.cfg.host, port=self.cfg.port, error=str(e))

102 raise OSError(f"Connection Failure to {self.cfg.host}:{self.cfg.port} as {self.cfg.user} -- {e}") from e

103

104 def stop(self) -> None:

105 if self.client:

106 self.client.loop_stop()

107 self.client.disconnect()

108 self.client = None

109 self.connected.clear()

110

111 def is_available(self) -> bool:

112 return self.client is not None and not self.fatal_failure.is_set()

113

114 def on_connect(

115 self, _client: mqtt.Client, _userdata: Any, _flags: mqtt.ConnectFlags, rc: ReasonCode, _props: Properties | None

116 ) -> None:

117 if not self.client or self.fatal_failure.is_set():

118 self.log.warn("No client, check if started and authorized")

119 return

120 if rc.getName() == "Not authorized":

121 self.fatal_failure.set()

122 log.error("Invalid MQTT credentials", result_code=rc)

123 return

124 if rc != 0:

125 self.log.warning("Connection failed to broker", result_code=rc)

126 else:

127 self.log.debug("Connected successfully to MQTT broker")

128 self.connected.set()

129 for topic, provider in self.providers_by_topic.items():

130 self.log.debug("(Re)subscribing", topic=topic, provider=provider.source_type)

131 self.client.subscribe(topic)

132

133 def on_disconnect(

134 self,

135 _client: mqtt.Client,

136 _userdata: Any,

137 _disconnect_flags: mqtt.DisconnectFlags,

138 rc: ReasonCode,

139 _props: Properties | None,

140 ) -> None:

141 self.connected.clear()

142 if rc == 0:

143 self.log.debug("Disconnected from broker", result_code=rc)

144 else:

145 self.log.warning("Disconnect failure from broker", result_code=rc)

146

147 async def clean_topics(

148 self, provider: ReleaseProvider, wait_time: int = 5, max_time: int = 120, initial: bool = False

149 ) -> None:

150 logger = self.log.bind(action="clean")

151

152 if self.fatal_failure.is_set():

153 return

154 try:

155 logger.info("Starting clean cycle, wait time: %s, max time: %s, initial: %s", wait_time, max_time, initial)

156 cutoff_time: float = time.time() + max_time

157 cleaner = mqtt.Client(

158 callback_api_version=CallbackAPIVersion.VERSION1,

159 client_id=f"updates2mqtt_clean_{self.node_cfg.name}",

160 clean_session=True,

161 )

162 results = {"cleaned": 0, "matched": 0, "discovered": 0, "last_timestamp": time.time()}

163 cleaner.username_pw_set(self.cfg.user, password=self.cfg.password)

164 cleaner.connect(host=self.cfg.host, port=self.cfg.port, keepalive=self.cfg.keepalive)

165

166 def cleanup(_client: mqtt.Client, _userdata: Any, msg: mqtt.MQTTMessage) -> None:

167 discovery: Discovery | None = None

168 if msg.topic.startswith(

169 f"{self.hass_cfg.discovery.prefix}/update/{self.node_cfg.name}_{provider.source_type}_"

170 ):

171 discovery = self.reverse_config_topic(msg.topic, provider.source_type)

172 elif msg.topic.startswith( 172 ↛ 175line 172 didn't jump to line 175 because the condition on line 172 was never true

173 f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/"

174 ) and msg.topic.endswith("/state"):

175 discovery = self.reverse_state_topic(msg.topic, provider.source_type)

176 elif msg.topic.startswith(f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/"): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 discovery = self.reverse_general_topic(msg.topic, provider.source_type)

178 else:

179 logger.debug("Ignoring other topic ", topic=msg.topic)

180 return

181 results["discovered"] += 1

182 if not initial and discovery is None:

183 logger.debug("Removing unknown discovery", topic=msg.topic)

184 cleaner.publish(msg.topic, "", retain=True)

185 results["cleaned"] += 1

186 elif discovery is not None: 186 ↛ 189line 186 didn't jump to line 189 because the condition on line 186 was always true

187 results["matched"] += 1

188

189 try:

190 if msg.payload: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true

191 payload = json.loads(msg.payload)

192 update_section = payload.get("update") if isinstance(payload.get("update"), dict) else None

193 lingering_in_progress = payload.get("in_progress") or (

194 update_section and update_section.get("in_progress")

195 )

196 if lingering_in_progress and initial:

197 logger.info("Clearing lingering in-progress state at %s", msg.topic)

198 payload["in_progress"] = False

199 if update_section is not None:

200 update_section["in_progress"] = False

201 cleaner.publish(msg.topic, json.dumps(payload), retain=True)

202 results["cleaned"] += 1

203 elif (

204 initial

205 and msg.topic.endswith("/state")

206 and (payload.get("installed_version") is None or payload.get("latest_version") is None)

207 ):

208 # Stale/incomplete state message (e.g. from an older schema) leaves HA showing

209 # "unknown" forever, since it has nothing to compare against. Clear it so the

210 # upcoming scan can publish a complete replacement.

211 logger.info("Clearing stale incomplete state at %s", msg.topic)

212 cleaner.publish(msg.topic, "", retain=True)

213 results["cleaned"] += 1

214 except Exception as e:

215 logger.warn("Invalid payload at %s: %s", msg.topic, e)

216 cleaner.publish(msg.topic, "", retain=True)

217 results["cleaned"] += 1

218

219 results["last_timestamp"] = time.time()

220

221 cleaner.on_message = cleanup

222 options = paho.mqtt.subscribeoptions.SubscribeOptions(noLocal=True)

223 cleaner.subscribe(f"{self.hass_cfg.discovery.prefix}/update/#", options=options)

224 cleaner.subscribe(f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}/#", options=options)

225

226 while time.time() - results["last_timestamp"] <= wait_time and time.time() <= cutoff_time:

227 cleaner.loop(0.5)

228

229 logger.info(

230 f"Cleaned - discovered:{results['discovered']}, matched:{results['matched']}, cleaned:{results['cleaned']}"

231 )

232 except Exception as e:

233 logger.exception("Cleaning topics of stale entries failed: %s", e)

234

235 def safe_json_decode(self, jsonish: str | bytes | None) -> dict:

236 if jsonish is None:

237 return {}

238 try:

239 return json.loads(jsonish)

240 except Exception:

241 log.exception("JSON decode fail (%s)", jsonish)

242 try:

243 return json.loads(jsonish[1:-1])

244 except Exception:

245 log.exception("JSON decode fail (%s)", jsonish[1:-1])

246 return {}

247

248 def validate_command(self, msg: MQTTMessage | LocalMessage) -> tuple[ReleaseProvider, str, str] | None:

249

250 logger = self.log.bind(topic=msg.topic, payload=msg.payload)

251 comp_name: str | None = None

252 command: str | None = None

253 try:

254 logger.info("Command received for %s", msg.topic)

255 source_type: str | None = None

256

257 payload: str | None = None

258 if isinstance(msg.payload, bytes):

259 payload = msg.payload.decode("utf-8")

260 elif isinstance(msg.payload, str): 260 ↛ 262line 260 didn't jump to line 262 because the condition on line 260 was always true

261 payload = msg.payload

262 if payload and "|" in payload:

263 source_type, comp_name, command = payload.split("|")

264 else:

265 logger.warn("Invalid command format, expecting `source_type|comp_name|command`")

266 return None

267 logger.debug("Validating %s:%s:%s", source_type, comp_name, command)

268

269 provider: ReleaseProvider | None = self.providers_by_topic.get(msg.topic) if msg.topic else None

270

271 if not provider: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true

272 logger.warn("Unexpected provider type %s", msg.topic)

273 return None

274 if source_type is None or provider.source_type != source_type:

275 logger.warn("Unexpected source type %s", source_type)

276 return None

277 if command != "install": 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true

278 logger.warn("Unknown command: %s", command)

279 return None

280 if not comp_name: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 logger.warn("Missing comp_name in command message: %s", msg.payload)

282 return None

283

284 in_progress_key: tuple[str, str] = (source_type, comp_name)

285 if in_progress_key in self.commands_in_progress:

286 logger.warn("Ignoring duplicate %s command for %s, already in progress", command, comp_name)

287 else:

288 self.commands_in_progress.add(in_progress_key)

289 return (provider, comp_name, command)

290 except Exception:

291 logger.error("Unexpected error validating command")

292 return None

293

294 async def execute_command(

295 self, provider: ReleaseProvider, comp_name: str, command: str, on_update_start: Callable, on_update_end: Callable

296 ) -> None:

297 # TODO: defer handling of commands where repository is throttled

298 logger = self.log.bind(source_type=provider.source_type, comp_name=comp_name, command=command)

299 try:

300 logger.info("Execution starting for %s %s", command, comp_name)

301

302 in_progress_key: tuple[str, str] = (provider.source_type, comp_name)

303 logger.info(

304 "Passing %s command to %s scanner for %s",

305 command,

306 provider.source_type,

307 comp_name,

308 )

309 try:

310 updated: bool = provider.command(comp_name, command, on_update_start, on_update_end)

311 discovery = provider.resolve(comp_name)

312 if updated and discovery: 312 ↛ 320line 312 didn't jump to line 320 because the condition on line 312 was always true

313 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT and self.hass_cfg.discovery.enabled:

314 self.publish_hass_config(discovery)

315 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):

316 self.publish_discovery(discovery)

317 if discovery and discovery.publish_policy == PublishPolicy.HOMEASSISTANT:

318 self.publish_hass_state(discovery)

319 else:

320 logger.debug("No change to republish after execution")

321 finally:

322 if in_progress_key in self.commands_in_progress: 322 ↛ 324line 322 didn't jump to line 324 because the condition on line 322 was always true

323 self.commands_in_progress.discard(in_progress_key)

324 logger.info("Execution ended")

325 except Exception:

326 logger.exception("Execution failed")

327

328 def local_message(self, discovery: Discovery, command: str) -> None:

329 """Simulate an incoming MQTT message for local commands"""

330 msg = LocalMessage(

331 topic=self.command_topic(discovery.provider), payload="|".join([discovery.source_type, discovery.name, command])

332 )

333 self.handle_message(msg)

334

335 def on_subscribe(

336 self,

337 _client: mqtt.Client,

338 userdata: Any,

339 mid: int,

340 reason_code_list: list[ReasonCode],

341 properties: Properties | None = None,

342 ) -> None:

343 self.log.debug(

344 "on_subscribe, userdata=%s, mid=%s, reasons=%s, properties=%s", userdata, mid, reason_code_list, properties

345 )

346

347 def on_unsubscribe(

348 self,

349 _client: mqtt.Client,

350 userdata: Any,

351 mid: int,

352 reason_code_list: list[ReasonCode],

353 properties: Properties | None = None,

354 ) -> None:

355 self.log.debug(

356 "on_unsubscribe, userdata=%s, mid=%s, reasons=%s, properties=%s", userdata, mid, reason_code_list, properties

357 )

358

359 def on_message(self, _client: mqtt.Client, _userdata: Any, msg: mqtt.MQTTMessage) -> None:

360 """Callback for incoming MQTT messages""" # noqa: D401

361 if msg.topic in self.providers_by_topic:

362 self.handle_message(msg)

363 else:

364 # apparently the root non-wildcard sub sometimes brings in child topics

365 self.log.debug("Unhandled message #%s on %s:%s", msg.mid, msg.topic, msg.payload)

366

367 def handle_message(self, msg: mqtt.MQTTMessage | LocalMessage) -> None:

368 def update_start(discovery: Discovery) -> None:

369 self.log.debug("on_update_start: %s", topic=msg.topic)

370 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT:

371 self.publish_hass_state(discovery, in_progress=True)

372 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):

373 self.publish_discovery(discovery, in_progress=True)

374

375 def update_end(discovery: Discovery) -> None:

376 self.log.debug("on_update_end: %s", topic=msg.topic)

377 if discovery.publish_policy == PublishPolicy.HOMEASSISTANT:

378 self.publish_hass_state(discovery, in_progress=False)

379 if discovery.publish_policy in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT):

380 self.publish_discovery(discovery, in_progress=False)

381

382 # TODO: fix double publish on callback and in command exec

383 if self.event_loop is not None: 383 ↛ 399line 383 didn't jump to line 399 because the condition on line 383 was always true

384 self.log.debug("Executing command topic", topic=msg.topic)

385 parsed: tuple[ReleaseProvider, str, str] | None = self.validate_command(msg=msg)

386 if parsed is not None: 386 ↛ exitline 386 didn't return from function 'handle_message' because the condition on line 386 was always true

387 provider, comp_name, command = parsed

388 asyncio.run_coroutine_threadsafe(

389 self.execute_command(

390 provider=provider,

391 comp_name=comp_name,

392 command=command,

393 on_update_start=update_start,

394 on_update_end=update_end,

395 ),

396 loop=self.event_loop,

397 )

398 else:

399 self.log.error("No event loop to handle message", topic=msg.topic)

400

401 def config_topic(self, discovery: Discovery) -> str:

402 prefix = self.hass_cfg.discovery.prefix

403 return f"{prefix}/update/{self.node_cfg.name}_{discovery.source_type}_{discovery.name}/update/config"

404

405 def reverse_config_topic(self, topic: str, source_type: str) -> Discovery | None:

406 match = re.fullmatch(

407 f"{self.hass_cfg.discovery.prefix}/update/{self.node_cfg.name}_{source_type}_({MQTT_NAME})/update/config",

408 topic,

409 )

410 if match and len(match.groups()) == 1: 410 ↛ 415line 410 didn't jump to line 415 because the condition on line 410 was always true

411 discovery_name: str = match.group(1)

412 if source_type in self.providers_by_type and discovery_name in self.providers_by_type[source_type].discoveries:

413 return self.providers_by_type[source_type].discoveries[discovery_name]

414

415 self.log.debug("MQTT CONFIG no match for %s", topic)

416 return None

417

418 def state_topic(self, discovery: Discovery) -> str:

419 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{discovery.source_type}/{discovery.name}/state"

420

421 def reverse_state_topic(self, topic: str, source_type: str) -> Discovery | None:

422 match = re.fullmatch(

423 f"{self.cfg.topic_root}/{self.node_cfg.name}/{source_type}/({MQTT_NAME})/state",

424 topic,

425 )

426 if match and len(match.groups()) == 1:

427 discovery_name: str = match.group(1)

428 if discovery_name in self.providers_by_type[source_type].discoveries:

429 return self.providers_by_type[source_type].discoveries[discovery_name]

430

431 self.log.debug("MQTT STATE no match for %s", topic)

432 return None

433

434 def general_topic(self, discovery: Discovery) -> str:

435 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{discovery.source_type}/{discovery.name}"

436

437 def reverse_general_topic(self, topic: str, source_type: str) -> Discovery | None:

438 match = re.fullmatch(f"{self.cfg.topic_root}/{self.node_cfg.name}/{source_type}/({MQTT_NAME})", topic)

439 if match and len(match.groups()) == 1:

440 discovery_name: str = match.group(1)

441 if discovery_name in self.providers_by_type[source_type].discoveries:

442 return self.providers_by_type[source_type].discoveries[discovery_name]

443

444 self.log.debug("MQTT ATTR no match for %s", topic)

445 return None

446

447 def command_topic(self, provider: ReleaseProvider) -> str:

448 return f"{self.cfg.topic_root}/{self.node_cfg.name}/{provider.source_type}"

449

450 def publish_discovery(self, discovery: Discovery, in_progress: bool = False) -> None:

451 """Comprehensive, non Home Assistant specific, base publication"""

452 if discovery.publish_policy not in (PublishPolicy.HOMEASSISTANT, PublishPolicy.MQTT): 452 ↛ 453line 452 didn't jump to line 453 because the condition on line 452 was never true

453 return

454 self.log.debug("Discovery publish: %s", discovery)

455 payload: dict[str, Any] = discovery.as_dict()

456 payload["update"]["in_progress"] = in_progress # ty:ignore[invalid-assignment]

457 if payload.get("release", {}).get("summary") and self.hass_cfg.release_summary_max_size: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true

458 payload["release"]["summary"] = payload["release"]["summary"][: self.hass_cfg.release_summary_max_size]

459 self.publish(self.general_topic(discovery), payload)

460

461 def publish_hass_state(self, discovery: Discovery, in_progress: bool = False) -> None:

462 if discovery.publish_policy != PublishPolicy.HOMEASSISTANT: 462 ↛ 463line 462 didn't jump to line 463 because the condition on line 462 was never true

463 return

464 self.log.debug("HASS State update, in progress: %s, discovery: %s", in_progress, discovery)

465 self.publish(

466 self.state_topic(discovery),

467 hass_format_state(

468 discovery, in_progress=in_progress, release_summary_max_size=self.hass_cfg.release_summary_max_size

469 ),

470 )

471

472 def publish_hass_config(self, discovery: Discovery) -> None:

473 if discovery.publish_policy != PublishPolicy.HOMEASSISTANT: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true

474 return

475 object_id = f"{discovery.source_type}_{self.node_cfg.name}_{discovery.name}"

476 self.log.debug("HASS Config: %s", object_id)

477

478 self.publish(

479 self.config_topic(discovery),

480 hass_format_config(

481 discovery=discovery,

482 object_id=object_id,

483 area=self.hass_cfg.area,

484 state_topic=self.state_topic(discovery),

485 attrs_topic=self.general_topic(discovery) if self.hass_cfg.extra_attributes else None,

486 command_topic=self.command_topic(discovery.provider),

487 force_command_topic=self.hass_cfg.force_command_topic,

488 device_creation=self.hass_cfg.device_creation,

489 ),

490 )

491

492 def subscribe_hass_command(self, provider: ReleaseProvider): # noqa: ANN201

493 topic = self.command_topic(provider)

494 if topic in self.providers_by_topic or self.client is None:

495 self.log.debug("Skipping subscription", topic=topic)

496 else:

497 self.log.info("Handler subscribing", topic=topic)

498 self.providers_by_topic[topic] = provider

499 self.providers_by_type[provider.source_type] = provider

500 self.client.subscribe(topic)

501 return topic

502

503 def loop_once(self) -> None:

504 if self.client:

505 self.client.loop()

506

507 def publish(self, topic: str, payload: dict, qos: int = 1, retain: bool = True) -> None:

508 if self.client:

509 info: MQTTMessageInfo = self.client.publish(topic, payload=json.dumps(payload), qos=qos, retain=retain)

510 if info.rc == MQTTErrorCode.MQTT_ERR_SUCCESS:

511 self.log.debug(

512 "Publish to %s, mid: %s, published: %s, qos: %s, rc: %s", topic, info.mid, info.is_published(), qos, info.rc

513 )

514 elif info.rc == MQTTErrorCode.MQTT_ERR_NO_CONN and qos > 0: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true

515 self.log.debug(

516 "Not currently connected, queued for delivery on reconnect: %s, mid: %s, qos: %s",

517 topic,

518 info.mid,

519 qos,

520 )

521 else:

522 self.log.warning("Problem publishing to %s, mid: %s, qos: %s, rc: %s", topic, info.mid, qos, info.rc)

523 else:

524 self.log.debug("No client to publish at %s", topic)