diff --git a/reactortest/src/main/java/component/api/MComponent.java b/reactortest/src/main/java/component/api/MComponent.java
index 1a9943b..7f7027c 100644
--- a/reactortest/src/main/java/component/api/MComponent.java
+++ b/reactortest/src/main/java/component/api/MComponent.java
@@ -1,11 +1,11 @@
 package component.api;
 
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 public interface MComponent<F> {
     <T> Publisher<T> from(String destination, Function<F, T> converter) throws Exception;
-    <T> Consumer<T> to(String destination, Function<T, F> converter) throws Exception;
+    <T> Subscriber<T> to(String destination, Function<T, F> converter) throws Exception;
 }
diff --git a/reactortest/src/main/java/component/eventadmin/EventAdminComponent.java b/reactortest/src/main/java/component/eventadmin/EventAdminComponent.java
index 6bb1b40..c2ce58f 100644
--- a/reactortest/src/main/java/component/eventadmin/EventAdminComponent.java
+++ b/reactortest/src/main/java/component/eventadmin/EventAdminComponent.java
@@ -1,7 +1,6 @@
 package component.eventadmin;
 
 import java.util.Map;
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -12,6 +11,7 @@
 import org.osgi.service.component.annotations.Reference;
 import org.osgi.service.event.EventAdmin;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import component.api.MComponent;
 
@@ -39,7 +39,7 @@
     }  
     
     @Override
-    public <T> Consumer<T> to(String topic, Function<T, Map<String, ?>> converter) throws Exception {
+    public <T> Subscriber<T> to(String topic, Function<T, Map<String, ?>> converter) throws Exception {
         return new EventAdminDestination<T>(client, topic, converter);
     }
 
diff --git a/reactortest/src/main/java/component/eventadmin/EventAdminDestination.java b/reactortest/src/main/java/component/eventadmin/EventAdminDestination.java
index 6144056..a1a8b96 100644
--- a/reactortest/src/main/java/component/eventadmin/EventAdminDestination.java
+++ b/reactortest/src/main/java/component/eventadmin/EventAdminDestination.java
@@ -1,17 +1,19 @@
 package component.eventadmin;
 
 import java.util.Map;
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.osgi.service.event.Event;
 import org.osgi.service.event.EventAdmin;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
-public class EventAdminDestination<T> implements Consumer<T> {
+public class EventAdminDestination<T> implements Subscriber<T> {
     
     private EventAdmin client;
     private String topic;
     private Function<T, Map<String, ?>> converter;
+    private Subscription subscription;
 
     public EventAdminDestination(EventAdmin client, String topic, Function<T, Map<String, ?>> converter) {
         this.client = client;
@@ -20,14 +22,29 @@
     }
 
     @Override
-    public void accept(T payload) {
+    public void onSubscribe(Subscription s) {
+        this.subscription = s;
+        s.request(1);
+    }
+
+    @Override
+    public void onNext(T payload) {
         try {
             Event event = new Event(topic, converter.apply(payload));
             this.client.sendEvent(event);
         } catch (Exception e) {
             throw new RuntimeException(e);
+        } finally {
+            subscription.request(1);
         }
     }
 
+    @Override
+    public void onError(Throwable t) {
+    }
+
+    @Override
+    public void onComplete() {
+    }
 
 }
diff --git a/reactortest/src/main/java/component/mail/MailComponent.java b/reactortest/src/main/java/component/mail/MailComponent.java
index 962b0b8..a932271 100644
--- a/reactortest/src/main/java/component/mail/MailComponent.java
+++ b/reactortest/src/main/java/component/mail/MailComponent.java
@@ -1,6 +1,5 @@
 package component.mail;
 
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import javax.mail.MethodNotSupportedException;
@@ -10,6 +9,7 @@
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Reference;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import component.api.MComponent;
 
@@ -25,7 +25,7 @@
     }  
     
     @Override
-    public <T> Consumer<T> to(String destination, Function<T, MimeMessage> converter) throws Exception {
+    public <T> Subscriber<T> to(String destination, Function<T, MimeMessage> converter) throws Exception {
         return new MailDestination<T>(destination, converter);
     }
 
diff --git a/reactortest/src/main/java/component/mail/MailDestination.java b/reactortest/src/main/java/component/mail/MailDestination.java
index 74cde05..33dd1da 100644
--- a/reactortest/src/main/java/component/mail/MailDestination.java
+++ b/reactortest/src/main/java/component/mail/MailDestination.java
@@ -1,6 +1,5 @@
 package component.mail;
 
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import javax.mail.Address;
@@ -8,10 +7,14 @@
 import javax.mail.internet.InternetAddress;
 import javax.mail.internet.MimeMessage;
 
-public class MailDestination<T> implements Consumer<T> {
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public class MailDestination<T> implements Subscriber<T> {
 
     private String destination;
     private Function<T, MimeMessage> converter;
+    private Subscription subscription;
 
     public MailDestination(String destination, Function<T, MimeMessage> converter) {
         this.destination = destination;
@@ -19,14 +22,30 @@
     }
 
     @Override
-    public void accept(T payload) {
+    public void onSubscribe(Subscription s) {
+        this.subscription = s;
+        s.request(1);
+    }
+
+    @Override
+    public void onNext(T payload) {
         try {
             MimeMessage message = converter.apply(payload);
             Address[] addresses = new Address[]{new InternetAddress(destination)};
             Transport.send(message, addresses);
         } catch (Exception e) {
             throw new RuntimeException(e);
+        } finally {
+            subscription.request(1);
         }
     }
 
+    @Override
+    public void onError(Throwable t) {
+    }
+
+    @Override
+    public void onComplete() {
+    }
+
 }
diff --git a/reactortest/src/main/java/component/mqtt/MqttComponent.java b/reactortest/src/main/java/component/mqtt/MqttComponent.java
index c5c6d08..a4c3b46 100644
--- a/reactortest/src/main/java/component/mqtt/MqttComponent.java
+++ b/reactortest/src/main/java/component/mqtt/MqttComponent.java
@@ -1,6 +1,5 @@
 package component.mqtt;
 
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -10,6 +9,7 @@
 import org.osgi.service.component.annotations.Deactivate;
 import org.osgi.service.metatype.annotations.ObjectClassDefinition;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import component.api.MComponent;
 
@@ -42,7 +42,7 @@
     }  
     
     @Override
-    public <T> Consumer<T> to(String topic, Function<T, byte[]> converter) throws Exception {
+    public <T> Subscriber<T> to(String topic, Function<T, byte[]> converter) throws Exception {
         return new MqttDestination<T>(client, topic, converter);
     }
 
diff --git a/reactortest/src/main/java/component/mqtt/MqttDestination.java b/reactortest/src/main/java/component/mqtt/MqttDestination.java
index dcca169..d5a874d 100644
--- a/reactortest/src/main/java/component/mqtt/MqttDestination.java
+++ b/reactortest/src/main/java/component/mqtt/MqttDestination.java
@@ -1,16 +1,18 @@
 package component.mqtt;
 
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
-public class MqttDestination<T> implements Consumer<T> {
+public class MqttDestination<T> implements Subscriber<T> {
     
     private MqttClient client;
     private String topic;
     private Function<T, byte[]> converter;
+    private Subscription subscription;
 
     public MqttDestination(MqttClient client, String topic, Function<T, byte[]> converter) {
         this.client = client;
@@ -19,14 +21,31 @@
     }
 
     @Override
-    public void accept(T payload) {
+    public void onSubscribe(Subscription s) {
+        this.subscription = s;
+        s.request(1);
+    }
+
+    @Override
+    public void onNext(T payload) {
         try {
             MqttMessage message = new MqttMessage(converter.apply(payload));
             this.client.publish(topic, message);
         } catch (Exception e) {
             throw new RuntimeException(e);
+        } finally {
+            subscription.request(1);
         }
     }
 
+    @Override
+    public void onError(Throwable t) {
+        System.out.println("onerr");
+    }
+
+    @Override
+    public void onComplete() {
+        System.out.println("oncomplete");
+    }
 
 }
diff --git a/reactortest/src/main/java/reactortest/ExampleEventAdmin.java b/reactortest/src/main/java/reactortest/ExampleEventAdmin.java
index 3291758..47ff7b7 100644
--- a/reactortest/src/main/java/reactortest/ExampleEventAdmin.java
+++ b/reactortest/src/main/java/reactortest/ExampleEventAdmin.java
@@ -1,12 +1,12 @@
 package reactortest;
 
 import java.util.Map;
-import java.util.function.Consumer;
 
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Reference;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -23,7 +23,7 @@
     @Activate
     public void start() throws Exception {
         Publisher<Map<String, ?>> fromTopic = eventAdmin.from("input", Map2Map::convert);
-        Consumer<Map<String, ?>> toTopic = eventAdmin.to("output", Map2Map::convert);
+        Subscriber<Map<String, ?>> toTopic = eventAdmin.to("output", Map2Map::convert);
         Flux.from(fromTopic)
             .log()
             .subscribe(toTopic);
diff --git a/reactortest/src/main/java/reactortest/MqttExampleComponent.java b/reactortest/src/main/java/reactortest/MqttExampleComponent.java
index a90e3fc..8d079bd 100644
--- a/reactortest/src/main/java/reactortest/MqttExampleComponent.java
+++ b/reactortest/src/main/java/reactortest/MqttExampleComponent.java
@@ -1,11 +1,10 @@
 package reactortest;
 
-import java.util.function.Consumer;
-
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Reference;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -24,7 +23,7 @@
     public void start() throws Exception {
         LOG.info("Starting mqtt test component");
         Publisher<Integer> fromTopic = mqtt.from("input", ByteArrayConverter::asInteger);
-        Consumer<Double> toTopic = mqtt.to("output", DoubleConverter::asByteAr);
+        Subscriber<Double> toTopic = mqtt.to("output", DoubleConverter::asByteAr);
         Flux.from(fromTopic)
             .log()
             .window(2, 1)
diff --git a/reactortest/src/test/java/component/mail/TestMail.java b/reactortest/src/test/java/component/mail/TestMail.java
index 16d9ea9..a37fec0 100644
--- a/reactortest/src/test/java/component/mail/TestMail.java
+++ b/reactortest/src/test/java/component/mail/TestMail.java
@@ -2,7 +2,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.function.Consumer;
 
 import javax.mail.MessagingException;
 import javax.mail.Session;
@@ -10,9 +9,8 @@
 
 import org.junit.Ignore;
 import org.junit.Test;
+import org.reactivestreams.Subscriber;
 
-import component.mail.MailComponent;
-import component.mail.SessionComponent;
 import reactor.core.publisher.Flux;
 
 public class TestMail {
@@ -27,7 +25,7 @@
         MailComponent mail = new MailComponent();
         mail.session = session;
 
-        Consumer<String> to = mail.to("cschneider@localhost", txt -> createMessage(session, txt));
+        Subscriber<String> to = mail.to("cschneider@localhost", txt -> createMessage(session, txt));
         Flux.just("Test").subscribe(to);
     }
 
diff --git a/reactortest/src/test/java/component/mqtt/Test1.java b/reactortest/src/test/java/component/mqtt/Test1.java
index f6a0dfa..2c15d1f 100644
--- a/reactortest/src/test/java/component/mqtt/Test1.java
+++ b/reactortest/src/test/java/component/mqtt/Test1.java
@@ -3,7 +3,8 @@
 import static java.time.Duration.of;
 
 import java.time.temporal.ChronoUnit;
-import java.util.function.Consumer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
 import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -11,6 +12,7 @@
 import org.junit.Assert;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import reactor.core.publisher.Flux;
 import reactor.math.MathFlux;
@@ -41,12 +43,13 @@
 
     @Test
     public void testMQtt() throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
         MqttClient client = new MqttClient("tcp://192.168.0.126:1883", MqttClient.generateClientId());
         client.connect();
         MqttComponent mqtt = new MqttComponent();
         mqtt.client = client;
         Publisher<Integer> fromTopic = mqtt.from("input", ByteArrayConverter::asInteger);
-        Consumer<Double> toTopic = mqtt.to("output", DoubleConverter::asByteAr);
+        Subscriber<Double> toTopic = mqtt.to("output", DoubleConverter::asByteAr);
         Flux.from(fromTopic)
             .log()
             .window(2, 1)
@@ -56,10 +59,11 @@
         
         client.subscribe("output", (topic, message) -> {
             result = ByteArrayConverter.asDouble(message.getPayload());
+            latch.countDown();
         });
         client.publish("input", new MqttMessage(ByteArrayConverter.fromInteger(2)));
         client.publish("input", new MqttMessage(new Integer(2).toString().getBytes()));
-        Thread.sleep(100);
+        latch.await(10, TimeUnit.SECONDS);
         Assert.assertEquals(2, result, 0.1);
         client.disconnect();
         client.close();