diff --git a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
index 9e8e1e10..6a4a4a67 100755
--- a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
+++ b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
@@ -73,7 +73,7 @@ class Processor(ConsumerProducer):
                 chunks=embeds,
             )
 
-            await self.producer.send(r)
+            await self.send(r)
 
         except Exception as e:
             print("Exception:", e, flush=True)
diff --git a/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py b/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py
index f76f3235..2cbe9907 100755
--- a/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py
+++ b/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py
@@ -77,7 +77,7 @@ class Processor(ConsumerProducer):
                 entities=entities,
             )
 
-            await self.producer.send(r)
+            await self.send(r)
 
         except Exception as e:
             print("Exception:", e, flush=True)
diff --git a/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py b/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py
index 62411cb5..c88005b7 100755
--- a/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py
+++ b/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py
@@ -110,7 +110,7 @@ class Processor(ConsumerProducer):
             metadata=metadata,
             entities=entities,
         )
-        await self.ec_prod.send(t)
+        self.ec_prod.send(t)
 
     async def handle(self, msg):