entrypoints

Author: martindurant

streamz/core.py

@@ -168,7 +168,7 @@ def wrapped(*args, **kwargs):
     def register_plugin_entry_point(cls, entry_point, modifier=identity):
         if hasattr(cls, entry_point.name):
             raise ValueError(
-                f"Can't add {entry_point.name} from {entry_point.module_name} "
+                f"Can't add {entry_point.name} "
                 f"to {cls.__name__}: duplicate method name."
             )
 
@@ -178,7 +178,6 @@ def stub(*args, **kwargs):
             if not issubclass(node, Stream):
                 raise TypeError(
                     f"Error loading {entry_point.name} "
-                    f"from module {entry_point.module_name}: "
                     f"{node.__class__.__name__} must be a subclass of Stream"
                 )
             if getattr(cls, entry_point.name).__name__ == "stub":

streamz/plugins.py

@@ -1,6 +1,6 @@
 import warnings
 
-import pkg_resources
+import importlib.metadata
 
 
 def try_register(cls, entry_point, *modifier):
@@ -14,9 +14,10 @@ def try_register(cls, entry_point, *modifier):
 
 
 def load_plugins(cls):
-    for entry_point in pkg_resources.iter_entry_points("streamz.sources"):
+    eps = importlib.metadata.entry_points()
+    for entry_point in eps.get("streamz.sources", []):
         try_register(cls, entry_point, staticmethod)
-    for entry_point in pkg_resources.iter_entry_points("streamz.nodes"):
+    for entry_point in eps.get("streamz.nodes", []):
         try_register(cls, entry_point)
-    for entry_point in pkg_resources.iter_entry_points("streamz.sinks"):
+    for entry_point in eps.get("streamz.sinks", []):
         try_register(cls, entry_point)

streamz/sources.py

@@ -896,12 +896,15 @@ class from_mqtt(from_q):
     :param client_kwargs:
         Passed to the client's ``connect()`` method
     """
-    def __init__(self, host, port, topic, keepalive=60 , client_kwargs=None, **kwargs):
+    def __init__(self, host, port, topic, keepalive=60 , client_kwargs=None,
+                 user=None, pw=None, **kwargs):
         self.host = host
         self.port = port
         self.keepalive = keepalive
         self.topic = topic
         self.client_kwargs = client_kwargs
+        self.user = user
+        self.pw = pw
         super().__init__(q=queue.Queue(), **kwargs)
 
     def _on_connect(self, client, userdata, flags, rc):
@@ -913,6 +916,8 @@ def _on_message(self, client, userdata, msg):
     async def run(self):
         import paho.mqtt.client as mqtt
         client = mqtt.Client()
+        if self.user:
+            client.username_pw_set(self.user, self.pw)
         client.on_connect = self._on_connect
         client.on_message = self._on_message
         client.connect(self.host, self.port, self.keepalive, **(self.client_kwargs or {}))

streamz/tests/test_kafka.py

@@ -51,7 +51,7 @@ def stop_docker(name='streamz-kafka', cid=None, let_fail=False):
 
 def launch_kafka():
     stop_docker(let_fail=True)
-    subprocess.call(shlex.split("docker pull spotify/kafka"))
+    subprocess.call(shlex.split("docker pull spotify/kafka"), stderr=subprocess.DEVNULL)
     cmd = ("docker run -d -p 2181:2181 -p 9092:9092 --env "
            "ADVERTISED_HOST=127.0.0.1 --env ADVERTISED_PORT=9092 "
            "--name streamz-kafka spotify/kafka")