Merge branch 'main' into 0.11.3-rc.1

.dockerignore

@@ -0,0 +1,7 @@
+docs/**
+docs_test/**
+docs_web/**
+tests/**
+htmlcov/**
+cache/**
+dist/**

docs/changelog.md

@@ -1,5 +1,31 @@
 # Changelog
 
+## 0.11.2
+
+-  config-file based plugin loading   [PR #240](https://github.com/Yakifo/amqtt/pull/240)
+-  dockerfile build update to support psutils   [PR #239](https://github.com/Yakifo/amqtt/pull/239)
+-  pass client session info to event callbacks   [PR #241](https://github.com/Yakifo/amqtt/pull/241)
+-  Require at least one auth   [PR #244](https://github.com/Yakifo/amqtt/pull/244)
+-  improvements in retaining messages   [PR #248](https://github.com/Yakifo/amqtt/pull/248)
+-  updating docker compose with resource limits   [PR #253](https://github.com/Yakifo/amqtt/pull/253)
+-  improve static type checking for plugin's `Config` class   [PR #249](https://github.com/Yakifo/amqtt/pull/249)
+-  broker shouldn't allow clients to publish to '$' topics   [PR #254](https://github.com/Yakifo/amqtt/pull/254)
+-  publishing to a topic with `*` is allowed, while `#` and `+` are not   [PR #251](https://github.com/Yakifo/amqtt/pull/251)
+-  updated samples; plugin config consistency (yaml and python dict)   [PR #252](https://github.com/Yakifo/amqtt/pull/252)
+-  add cpu, mem and broker version to dashboard   [PR #257](https://github.com/Yakifo/amqtt/pull/257)
+- [Issue 246](https://github.com/Yakifo/amqtt/issues/246) don't retain QoS 1 or 2 messages if client connects with clean session true
+- [Issue 175](https://github.com/Yakifo/amqtt/issues/175) plugin examples
+- [Issue 81](https://github.com/Yakifo/amqtt/issues/81) Abstract factory for plugins
+- [Issue 74](https://github.com/Yakifo/amqtt/issues/74) 模拟500个客户端并发，连接broker。
+- [Issue 60](https://github.com/Yakifo/amqtt/issues/60) amqtt server not relaying traffic
+- [Issue 31](https://github.com/Yakifo/amqtt/issues/31) Plugin config in yaml file not under - plugins entry
+- [Issue 27](https://github.com/Yakifo/amqtt/issues/27) don't retain messages from anonymous clients
+- [Issue 250](https://github.com/Yakifo/amqtt/issues/250) client doesn't prevent publishing to wildcard topics
+- [Issue 245](https://github.com/Yakifo/amqtt/issues/245) prevent clients from publishing to `$` topics
+- [Issue 196](https://github.com/Yakifo/amqtt/issues/196) proposal: enhancement to broker plugin configuration
+- [Issue 187](https://github.com/Yakifo/amqtt/issues/187) anonymous login allowed even if plugin isn't enabled
+- [Issue 123](https://github.com/Yakifo/amqtt/issues/123) Messages sent to mqtt can be consumed in time, but they occupy more and more memory
+
 ## 0.11.1
 
 - [PR #226](https://github.com/Yakifo/amqtt/pull/226) Consolidate super classes for plugins

docs/custom_plugins.md

@@ -1,14 +1,15 @@
-from dataclasses import dataclass
-
 # Custom Plugins
 
 With the aMQTT plugins framework, one can add additional functionality to the client or broker without
-having to rewrite any of the core logic.
+having to rewrite any of the core logic. Plugins can receive broker or client events [events](custom_plugins.md#events),
+used for [client authentication](custom_plugins.md#authentication-plugins) and controlling [topic access](custom_plugins.md#topic-filter-plugins).
+
+## Overview
 
 To create a custom plugin, subclass from `BasePlugin` (client or broker) or `BaseAuthPlugin` (broker only)
 or `BaseTopicPlugin` (broker only).  Each custom plugin may define settings specific to itself by creating
-a nested (or inner) `dataclass` named `Config` which declares each option and a default value (if applicable). A
-plugin's configuration dataclass will be type-checked and made available from within the `self.context` instance variable.
+a nested (ie. inner) `dataclass` named `Config` which declares each option and a default value (if applicable). A
+plugin's configuration dataclass will be type-checked and made available from within the `self.config` instance variable.
 
 ```python
 from dataclasses import dataclass, field
@@ -24,27 +25,44 @@ class TwoClassName(BasePlugin[BaseContext]):
     """This is a plugin with configuration options."""
     def __init__(self, context: BaseContext):
         super().__init__(context)
-        my_option_one: str = self.context.config.option1
+        self.my_option_one: str = self.config.option1
+        
+    async def on_broker_pre_start(self) -> None:
+        print(f"On broker pre-start, my option1 is: {self.my_option_one}")
         
     @dataclass
     class Config:
         option1: int
         option3: str = field(default="my_default_value")
-
 ```
 
 This plugin class then should be added to the configuration file of the broker or client (or to the `config`
-dictionary passed to the `Broker` or `MQTTClient`). 
+dictionary passed to the `Broker` or `MQTTClient`), such as `myBroker.yaml`:
 
 ```yaml
-...
-...
+---
+listeners:
+  default:
+    type: tcp
+    bind: 0.0.0.0:1883
 plugins:
   module.submodule.file.OneClassName:
   module.submodule.file.TwoClassName:
     option1: 123
 ```
 
+and then run via `amqtt -c myBroker.yaml`.
+
+??? note "Example: custom plugin within broker script"
+    The example `samples/broker_custom_plugin.py` demonstrates how to load a custom plugin
+    by passing a config dictionary when instantiating a `Broker`. While this example is functional,
+    `samples` is an invalid python module (it does not have a `__init__.py`); it is recommended
+    that custom plugins are placed in a python module.
+
+    ```python
+    --8<-- "samples/broker_custom_plugin.py"
+    ```
+
 ??? warning "Deprecated: activating plugins using `EntryPoints`"
     With the aMQTT plugins framework, one can add additional functionality to the client or broker without
     having to rewrite any of the core logic. To define a custom list of plugins to be loaded, add this section
@@ -60,8 +78,11 @@ plugins:
 
 ::: amqtt.plugins.base.BasePlugin
 
+
+
 ## Events
 
+
 All plugins are notified of events if the `BasePlugin` subclass implements one or more of these methods:
 
 ### Client and Broker

docs/references/broker_config.md

@@ -116,9 +116,9 @@ listeners:
         ssl: on
         cafile: /some/cafile
         capath: /some/folder
-        capath: certificate data
+        capath: 'certificate data'
         certfile: /some/certfile
-        keyfile: /some/key
+        keyfile: /some/keyfile
     my-ws-1:
         bind: 0.0.0.0:8080
         type: ws
@@ -127,7 +127,7 @@ listeners:
         type: ws
         ssl: on
         certfile: /some/certfile
-        keyfile: /some/key
+        keyfile: /some/keyfile
 timeout-disconnect-delay: 2
 plugins:
   - amqtt.plugins.authentication.AnonymousAuthPlugin:
@@ -137,7 +137,7 @@ plugins:
   - amqtt.plugins.topic_checking.TopicAccessControlListPlugin:
       acl:
         username1: ['repositories/+/master', 'calendar/#', 'data/memes']
-        username2: [ 'calendar/2025/#', 'data/memes']
+        username2: ['calendar/2025/#', 'data/memes']
         anonymous: ['calendar/2025/#']
 ```
 

docs_test/src/assets/helpers.tsx

@@ -1,7 +1,8 @@
 import React from "react";
 
 export type DataPoint = {
-  timestamp: string; // ISO format
+  time: string // ISO format
+  timestamp: number; // epoch milliseconds
   value: number;
 };
 

docs_test/src/dashboard/components/MainGrid.tsx

@@ -95,7 +95,8 @@ export default function MainGrid() {
         if(payload.topic in topic_map) {
           const { update } = topic_map[payload.topic];
           const newPoint: DataPoint = {
-            timestamp: new Date().toISOString(),
+            time: new Date().toISOString(),
+            timestamp: Date.now(),
             value: d
           };
           update(current => [...current, newPoint])
@@ -228,10 +229,10 @@ export default function MainGrid() {
         <strong>up for</strong> {serverUptime}
       </Grid>
         <Grid size={{xs: 12, md: 6}}>
-          <SessionsChart title={'Sent Messages'} label={''} data={sent} isConnected={isConnected}/>
+          <SessionsChart title={'Sent Messages'} label={''} data={sent} isConnected={isConnected} isPerSecond/>
         </Grid>
         <Grid size={{xs: 12, md: 6}}>
-          <SessionsChart title={'Received Messages'} label={''} data={received} isConnected={isConnected}/>
+          <SessionsChart title={'Received Messages'} label={''} data={received} isConnected={isConnected} isPerSecond/>
         </Grid>
         <Grid size={{xs: 12, md: 6}}>
           <SessionsChart title={'Bytes Out'} label={'Bytes'} data={bytesOut} isConnected={isConnected}/>

docs_test/src/dashboard/components/SessionsChart.tsx

@@ -7,6 +7,7 @@
   import CountUp from 'react-countup';
   import type { DataPoint } from '../../assets/helpers.jsx';
   import {CircularProgress} from "@mui/material";
+  import {useRef} from "react";
 
   const currentTimeZone = Intl.DateTimeFormat().resolvedOptions().timeZone;
 
@@ -117,6 +118,23 @@
 
   export default function SessionsChart(props: any) {
 
+    const lastCalc = useRef<number>(0);
+
+    const calc_per_second = (curValue: DataPoint, lastValue: DataPoint) => {
+      if(!props.isPerSecond) { return ''; }
+
+      if(!curValue || !lastValue) {
+        return '';
+      }
+
+      if(curValue.timestamp - lastValue.timestamp > 0) {
+        const per_second =  (curValue.value - lastValue.value) / ((curValue.timestamp - lastValue.timestamp) / 1000);
+        lastCalc.current = Math.trunc(per_second * 10) / 10;
+      }
+
+      return `${lastCalc.current} / sec`;
+    }
+
     return (
       <Card variant="outlined" sx={{ width: '100%' }}>
         <CardContent>
@@ -143,6 +161,9 @@
 
                 />} {props.label}
               </Typography>
+              <p>
+                { calc_per_second(props.data[props.data.length-1], props.data[props.data.length-2]) }
+              </p>
             </Stack>
           </Stack>
           { props.data.length < 2 ? <NoDataDisplay isConnected={props.isConnected}/> :

mkdocs.rtd.yml

@@ -136,7 +136,7 @@ plugins:
               ignore_init_summary: true
             docstring_section_style: list
             filters: ["!^_"]
-            heading_level: 1
+            heading_level: 2
             inherited_members: true
             merge_init_into_class: true
             parameter_headings: true

samples/broker_custom_plugin.py

@@ -0,0 +1,85 @@
+import asyncio
+import logging
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from amqtt.broker import Broker
+from amqtt.plugins.base import BasePlugin
+from amqtt.session import Session
+
+"""
+This sample shows how to run a broker without stacktraces on keyboard interrupt
+"""
+
+logger = logging.getLogger(__name__)
+
+
+class RemoteInfoPlugin(BasePlugin):
+
+    async def on_broker_client_connected(self, *, client_id:str, client_session:Session) -> None:
+        display_port_str = f"on port '{client_session.remote_port}'" if self.config.display_port else ''
+
+        logger.info(f"client '{client_id}' connected from"
+                    f" '{client_session.remote_address}' {display_port_str}")
+
+    @dataclass
+    class Config:
+        display_port: bool = False
+
+config = {
+    "listeners": {
+        "default": {
+            "type": "tcp",
+            "bind": "0.0.0.0:1883",
+        },
+        "ws-mqtt": {
+            "bind": "127.0.0.1:8080",
+            "type": "ws",
+            "max_connections": 10,
+        },
+    },
+    "plugins": {
+        'amqtt.plugins.authentication.AnonymousAuthPlugin': { 'allow_anonymous': True},
+        'samples.broker_custom_plugin.RemoteInfoPlugin': { 'display_port': True },
+    }
+}
+
+async def main_loop():
+    broker = Broker(config)
+    try:
+        await broker.start()
+        while True:
+            await asyncio.sleep(1)
+    except asyncio.CancelledError:
+        await broker.shutdown()
+
+async def main():
+    t = asyncio.create_task(main_loop())
+    try:
+        await t
+    except asyncio.CancelledError:
+        pass
+
+def __main__():
+
+    formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
+    logging.basicConfig(level=logging.INFO, format=formatter)
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    task = loop.create_task(main())
+
+    try:
+        loop.run_until_complete(task)
+    except KeyboardInterrupt:
+        logger.info("KeyboardInterrupt received. Stopping server...")
+        task.cancel()
+        loop.run_until_complete(task)  # Ensure task finishes cleanup
+    finally:
+        logger.info("Server stopped.")
+        loop.close()
+
+if __name__ == "__main__":
+    __main__()