diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfControllerImpl.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfControllerImpl.java
index 7477238..cdcfbf8 100644
--- a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfControllerImpl.java
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfControllerImpl.java
@@ -26,11 +26,11 @@
 import org.onosproject.netconf.NetconfDevice;
 import org.onosproject.netconf.NetconfDeviceInfo;
 import org.onosproject.netconf.NetconfDeviceListener;
+import org.onosproject.netconf.NetconfException;
 import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -90,9 +90,9 @@
     }
 
     @Override
-    public NetconfDevice connectDevice(NetconfDeviceInfo deviceInfo) throws IOException {
+    public NetconfDevice connectDevice(NetconfDeviceInfo deviceInfo) throws NetconfException {
         if (netconfDeviceMap.containsKey(deviceInfo.getDeviceId())) {
-            log.warn("Device {} is already present", deviceInfo);
+            log.info("Device {} is already present", deviceInfo);
             return netconfDeviceMap.get(deviceInfo.getDeviceId());
         } else {
             log.info("Creating NETCONF device {}", deviceInfo);
@@ -109,9 +109,8 @@
         }
     }
 
-    private NetconfDevice createDevice(NetconfDeviceInfo deviceInfo) throws IOException {
-        NetconfDevice netconfDevice = null;
-        netconfDevice = new NetconfDeviceImpl(deviceInfo);
+    private NetconfDevice createDevice(NetconfDeviceInfo deviceInfo) throws NetconfException {
+        NetconfDevice netconfDevice = new NetconfDeviceImpl(deviceInfo);
         for (NetconfDeviceListener l : netconfDeviceListeners) {
             l.deviceAdded(deviceInfo);
         }
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceImpl.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceImpl.java
index 762c21c..5a8f499 100644
--- a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceImpl.java
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceImpl.java
@@ -18,6 +18,7 @@
 
 import org.onosproject.netconf.NetconfDevice;
 import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfException;
 import org.onosproject.netconf.NetconfSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,12 +37,13 @@
     private boolean deviceState = false;
     private NetconfSession netconfSession;
 
-    public NetconfDeviceImpl(NetconfDeviceInfo deviceInfo) throws IOException {
+    public NetconfDeviceImpl(NetconfDeviceInfo deviceInfo) throws NetconfException {
         netconfDeviceInfo = deviceInfo;
         try {
             netconfSession = new NetconfSessionImpl(netconfDeviceInfo);
         } catch (IOException e) {
-            throw new IOException("Cannot create connection and session", e);
+            throw new NetconfException("Cannot create connection and session for device " +
+                                               deviceInfo, e);
         }
         deviceState = true;
     }
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceOutputEventListenerImpl.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceOutputEventListenerImpl.java
new file mode 100644
index 0000000..ce0c235
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfDeviceOutputEventListenerImpl.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.netconf.ctl;
+
+import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfDeviceOutputEvent;
+import org.onosproject.netconf.NetconfDeviceOutputEventListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Example of a listener for events that happen a Netconf session established
+ * for a particular NETCONF device.
+ */
+public class NetconfDeviceOutputEventListenerImpl implements NetconfDeviceOutputEventListener {
+
+    private static final Logger log =
+            LoggerFactory.getLogger(NetconfDeviceOutputEventListenerImpl.class);
+
+    private NetconfDeviceInfo deviceInfo;
+
+    public NetconfDeviceOutputEventListenerImpl(NetconfDeviceInfo deviceInfo) {
+        this.deviceInfo = deviceInfo;
+    }
+
+    @Override
+    public void event(NetconfDeviceOutputEvent event) {
+        switch (event.type()) {
+            case DEVICE_REPLY:
+                log.debug("Device {} has reply: {}", deviceInfo, event.getMessagePayload());
+                break;
+            case DEVICE_NOTIFICATION:
+                log.info("Device {} has notification: {}", deviceInfo, event.getMessagePayload());
+                break;
+            case DEVICE_UNREGISTERED:
+                log.warn("Device {} has closed session", deviceInfo);
+                //TODO tell onos about closed session
+                break;
+            case DEVICE_ERROR:
+                log.warn("Device {} has error: {}", deviceInfo, event.getMessagePayload());
+                break;
+            default:
+                log.warn("Wrong event type {} ", event.type());
+        }
+
+    }
+
+    @Override
+    public boolean isRelevant(NetconfDeviceOutputEvent event) {
+        return deviceInfo.equals(event.getDeviceInfo());
+    }
+}
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionDelegate.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionDelegate.java
new file mode 100644
index 0000000..1b526db
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionDelegate.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.netconf.ctl;
+
+import org.onosproject.netconf.NetconfDeviceOutputEvent;
+
+/**
+ * Entity associated with a NetconfSessionImpl and capable of receiving notifications of
+ * events about the session.
+ */
+public interface NetconfSessionDelegate {
+
+        /**
+         * Notifies the delegate via the specified event.
+         *
+         * @param event store generated event
+         */
+        void notify(NetconfDeviceOutputEvent event);
+}
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionImpl.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionImpl.java
index 795211e..7f544dc 100644
--- a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionImpl.java
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionImpl.java
@@ -18,56 +18,70 @@
 
 import ch.ethz.ssh2.Connection;
 import ch.ethz.ssh2.Session;
-import ch.ethz.ssh2.StreamGobbler;
 import com.google.common.base.Preconditions;
 import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfDeviceOutputEvent;
+import org.onosproject.netconf.NetconfDeviceOutputEventListener;
+import org.onosproject.netconf.NetconfException;
 import org.onosproject.netconf.NetconfSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.InputStreamReader;
 import java.io.PrintWriter;
-import java.io.StringWriter;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
 
 /**
  * Implementation of a NETCONF session to talk to a device.
  */
 public class NetconfSessionImpl implements NetconfSession {
 
-    public static final Logger log = LoggerFactory
+    private static final Logger log = LoggerFactory
             .getLogger(NetconfSessionImpl.class);
+
+
     private static final int CONNECTION_TIMEOUT = 0;
+    private static final String ENDPATTERN = "]]>]]>";
+    private static final AtomicInteger MESSAGE_ID_INTEGER = new AtomicInteger(0);
+    private static final String MESSAGE_ID_STRING = "message-id";
+    private static final String HELLO = "hello";
+    private static final String NEW_LINE = "\n";
 
 
     private Connection netconfConnection;
     private NetconfDeviceInfo deviceInfo;
     private Session sshSession;
     private boolean connectionActive;
-    private BufferedReader bufferReader = null;
     private PrintWriter out = null;
-    private int messageID = 0;
-    //TODO inject these capabilites from yang model provided by app
     private List<String> deviceCapabilities =
             Collections.singletonList("urn:ietf:params:netconf:base:1.0");
     private String serverCapabilities;
-    private String endpattern = "]]>]]>";
+    private NetconfStreamHandler t;
+    private Map<Integer, CompletableFuture<String>> replies;
 
 
-    public NetconfSessionImpl(NetconfDeviceInfo deviceInfo) throws IOException {
+    public NetconfSessionImpl(NetconfDeviceInfo deviceInfo) throws NetconfException {
         this.deviceInfo = deviceInfo;
         connectionActive = false;
+        replies = new HashMap<>();
         startConnection();
     }
 
 
-    private void startConnection() throws IOException {
+    private void startConnection() throws NetconfException {
         if (!connectionActive) {
             netconfConnection = new Connection(deviceInfo.ip().toString(), deviceInfo.port());
-            netconfConnection.connect(null, CONNECTION_TIMEOUT, 0);
+            try {
+                netconfConnection.connect(null, CONNECTION_TIMEOUT, 5000);
+            } catch (IOException e) {
+                throw new NetconfException("Cannot open a connection with device" + deviceInfo, e);
+            }
             boolean isAuthenticated;
             try {
                 if (deviceInfo.getKeyFile() != null) {
@@ -75,39 +89,49 @@
                             deviceInfo.name(), deviceInfo.getKeyFile(),
                             deviceInfo.password());
                 } else {
-                    log.info("authenticate with username {} and password {}",
-                             deviceInfo.name(), deviceInfo.password());
+                    log.debug("Authenticating to device {} with username {}",
+                              deviceInfo.getDeviceId(), deviceInfo.name(), deviceInfo.password());
                     isAuthenticated = netconfConnection.authenticateWithPassword(
                             deviceInfo.name(), deviceInfo.password());
                 }
             } catch (IOException e) {
-                throw new IOException("Authentication connection failed:" +
-                                              e.getMessage());
+                log.error("Authentication connection to device " +
+                                  deviceInfo.getDeviceId() + " failed:" +
+                                  e.getMessage());
+                throw new NetconfException("Authentication connection to device " +
+                                                   deviceInfo.getDeviceId() + " failed", e);
             }
 
             connectionActive = true;
             Preconditions.checkArgument(isAuthenticated,
-                                        "Authentication password and username failed");
+                                        "Authentication to device {} with username " +
+                                                "{} Failed",
+                                        deviceInfo.getDeviceId(), deviceInfo.name(),
+                                        deviceInfo.password());
             startSshSession();
         }
     }
 
-    private void startSshSession() throws IOException {
+    private void startSshSession() throws NetconfException {
         try {
             sshSession = netconfConnection.openSession();
             sshSession.startSubSystem("netconf");
-            bufferReader = new BufferedReader(new InputStreamReader(new StreamGobbler(
-                    sshSession.getStdout())));
             out = new PrintWriter(sshSession.getStdin());
+            t = new NetconfStreamThread(sshSession.getStdout(), sshSession.getStdin(),
+                                        sshSession.getStderr(), deviceInfo,
+                                        new NetconfSessionDelegateImpl());
+            this.addDeviceOutputListener(new NetconfDeviceOutputEventListenerImpl(deviceInfo));
             sendHello();
         } catch (IOException e) {
-            throw new IOException("Failed to create ch.ethz.ssh2.Session session:" +
-                                          e.getMessage());
+            log.error("Failed to create ch.ethz.ssh2.Session session:" +
+                              e.getMessage());
+            throw new NetconfException("Failed to create ch.ethz.ssh2.Session session with device" +
+                                               deviceInfo, e);
         }
     }
 
     private void sendHello() throws IOException {
-        serverCapabilities = doRequest(createHelloString());
+        serverCapabilities = sendRequest(createHelloString());
     }
 
     private String createHelloString() {
@@ -119,58 +143,68 @@
                 cap -> hellobuffer.append("    <capability>" + cap + "</capability>\n"));
         hellobuffer.append("  </capabilities>\n");
         hellobuffer.append("</hello>\n");
-        hellobuffer.append(endpattern);
+        hellobuffer.append(ENDPATTERN);
         return hellobuffer.toString();
 
     }
 
-    @Override
-    public String doRPC(String request) throws IOException {
-        String reply = doRequest(request + "\n" + endpattern);
-        return checkReply(reply) ? reply : "ERROR " + reply;
-    }
-
-    private String doRequest(String request) throws IOException {
-        //log.info("sshState " + sshSession.getState() + "request" + request);
-        checkAndRestablishSession();
-        //log.info("sshState after" + sshSession.getState());
-        out.print(request);
-        out.flush();
-        messageID++;
-        return readOne();
-    }
-
-    private void checkAndRestablishSession() throws IOException {
+    private void checkAndRestablishSession() throws NetconfException {
         if (sshSession.getState() != 2) {
             try {
                 startSshSession();
             } catch (IOException e) {
-                log.info("the connection had to be reopened");
+                log.debug("The connection with {} had to be reopened", deviceInfo.getDeviceId());
                 try {
                     startConnection();
                 } catch (IOException e2) {
                     log.error("No connection {} for device, exception {}", netconfConnection, e2);
-                    throw new IOException(e.getMessage());
-                    //TODO remove device from ONOS
+                    throw new NetconfException("Cannot re-open the connection with device" + deviceInfo, e);
                 }
             }
         }
     }
 
     @Override
-    public String get(String request) throws IOException {
-        return doRPC(request);
+    public String requestSync(String request) throws NetconfException {
+        String reply = sendRequest(request + NEW_LINE + ENDPATTERN);
+        return checkReply(reply) ? reply : "ERROR " + reply;
     }
 
     @Override
-    public String getConfig(String targetConfiguration) throws IOException {
+    public CompletableFuture<String> request(String request) {
+        CompletableFuture<String> ftrep = t.sendMessage(request);
+        replies.put(MESSAGE_ID_INTEGER.get(), ftrep);
+        return ftrep;
+    }
+
+    private String sendRequest(String request) throws NetconfException {
+        checkAndRestablishSession();
+        //FIXME find out a better way to enforce the presence of message-id
+        if (!request.contains(MESSAGE_ID_STRING) && !request.contains(HELLO)) {
+            request = request.replaceFirst("\">", "\" message-id=\""
+                    + MESSAGE_ID_INTEGER.get() + "\"" + ">");
+        }
+        CompletableFuture<String> futureReply = request(request);
+        MESSAGE_ID_INTEGER.incrementAndGet();
+        String rp = futureReply.join();
+        log.debug("Reply from device {}", rp);
+        return rp;
+    }
+
+    @Override
+    public String get(String request) throws NetconfException {
+        return requestSync(request);
+    }
+
+    @Override
+    public String getConfig(String targetConfiguration) throws NetconfException {
         return getConfig(targetConfiguration, null);
     }
 
     @Override
-    public String getConfig(String targetConfiguration, String configurationSchema) throws IOException {
+    public String getConfig(String targetConfiguration, String configurationSchema) throws NetconfException {
         StringBuilder rpc = new StringBuilder("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
-        rpc.append("<rpc message-id=\"" + messageID + "\"  "
+        rpc.append("<rpc message-id=\"" + MESSAGE_ID_INTEGER.get() + "\"  "
                            + "xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
         rpc.append("<get-config>\n");
         rpc.append("<source>\n");
@@ -183,23 +217,23 @@
         }
         rpc.append("</get-config>\n");
         rpc.append("</rpc>\n");
-        rpc.append(endpattern);
-        String reply = doRequest(rpc.toString());
+        rpc.append(ENDPATTERN);
+        String reply = sendRequest(rpc.toString());
         return checkReply(reply) ? reply : "ERROR " + reply;
     }
 
     @Override
-    public boolean editConfig(String newConfiguration) throws IOException {
-        newConfiguration = newConfiguration + endpattern;
-        return checkReply(doRequest(newConfiguration));
+    public boolean editConfig(String newConfiguration) throws NetconfException {
+        newConfiguration = newConfiguration + ENDPATTERN;
+        return checkReply(sendRequest(newConfiguration));
     }
 
     @Override
     public boolean editConfig(String targetConfiguration, String mode, String newConfiguration)
-            throws IOException {
+            throws NetconfException {
         newConfiguration = newConfiguration.trim();
         StringBuilder rpc = new StringBuilder("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
-        rpc.append("<rpc message-id=\"" + messageID + "\"  "
+        rpc.append("<rpc message-id=\"" + MESSAGE_ID_INTEGER.get() + "\"  "
                            + "xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
         rpc.append("<edit-config>");
         rpc.append("<target>");
@@ -213,13 +247,13 @@
         rpc.append("</config>");
         rpc.append("</edit-config>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString()));
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
     }
 
     @Override
     public boolean copyConfig(String targetConfiguration, String newConfiguration)
-            throws IOException {
+            throws NetconfException {
         newConfiguration = newConfiguration.trim();
         if (!newConfiguration.startsWith("<configuration>")) {
             newConfiguration = "<configuration>" + newConfiguration
@@ -237,12 +271,12 @@
         rpc.append("</source>");
         rpc.append("</copy-config>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString()));
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
     }
 
     @Override
-    public boolean deleteConfig(String targetConfiguration) throws IOException {
+    public boolean deleteConfig(String targetConfiguration) throws NetconfException {
         if (targetConfiguration.equals("running")) {
             log.warn("Target configuration for delete operation can't be \"running\"",
                      targetConfiguration);
@@ -257,12 +291,12 @@
         rpc.append("</target>");
         rpc.append("</delete-config>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString()));
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
     }
 
     @Override
-    public boolean lock() throws IOException {
+    public boolean lock() throws NetconfException {
         StringBuilder rpc = new StringBuilder("<?xml version=\"1.0\" " +
                                                       "encoding=\"UTF-8\"?>");
         rpc.append("<rpc>");
@@ -272,12 +306,12 @@
         rpc.append("</target>");
         rpc.append("</lock>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString()));
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
     }
 
     @Override
-    public boolean unlock() throws IOException {
+    public boolean unlock() throws NetconfException {
         StringBuilder rpc = new StringBuilder("<?xml version=\"1.0\" " +
                                                       "encoding=\"UTF-8\"?>");
         rpc.append("<rpc>");
@@ -287,16 +321,16 @@
         rpc.append("</target>");
         rpc.append("</unlock>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString()));
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
     }
 
     @Override
-    public boolean close() throws IOException {
+    public boolean close() throws NetconfException {
         return close(false);
     }
 
-    private boolean close(boolean force) throws IOException {
+    private boolean close(boolean force) throws NetconfException {
         StringBuilder rpc = new StringBuilder();
         rpc.append("<rpc>");
         if (force) {
@@ -306,8 +340,8 @@
         }
         rpc.append("<close-configuration/>");
         rpc.append("</rpc>");
-        rpc.append(endpattern);
-        return checkReply(doRequest(rpc.toString())) || close(true);
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString())) || close(true);
     }
 
     @Override
@@ -335,7 +369,17 @@
         deviceCapabilities = capabilities;
     }
 
-    private boolean checkReply(String reply) {
+    @Override
+    public void addDeviceOutputListener(NetconfDeviceOutputEventListener listener) {
+        t.addDeviceEventListener(listener);
+    }
+
+    @Override
+    public void removeDeviceOutputListener(NetconfDeviceOutputEventListener listener) {
+        t.removeDeviceEventListener(listener);
+    }
+
+    private boolean checkReply(String reply) throws NetconfException {
         if (reply != null) {
             if (!reply.contains("<rpc-error>")) {
                 return true;
@@ -345,36 +389,18 @@
                 return true;
             }
         }
-        log.warn("Error in reply {}", reply);
+        log.warn("Device " + deviceInfo + "has error in reply {}", reply);
         return false;
     }
 
-    private String readOne() throws IOException {
-        //TODO try a simple string
-        final StringWriter reply = new StringWriter();
-        while (true) {
-            int charRead = bufferReader.read();
-            if (charRead == -1) {
-                throw new IOException("Session closed");
-            }
+    public class NetconfSessionDelegateImpl implements NetconfSessionDelegate {
 
-            for (int i = 0; i < endpattern.length(); i++) {
-                if (charRead == endpattern.charAt(i)) {
-                    if (i < endpattern.length() - 1) {
-                        charRead = bufferReader.read();
-                    } else {
-                        return reply.getBuffer().toString();
-                    }
-                } else {
-                    String s = endpattern.substring(0, i);
-                    for (int j = 0; i < s.length(); j++) {
-                        reply.write(s.charAt(j));
-                    }
-                    reply.write(charRead);
-                    break;
-                }
-            }
+        @Override
+        public void notify(NetconfDeviceOutputEvent event) {
+            CompletableFuture<String> completedReply = replies.get(event.getMessageID());
+            completedReply.complete(event.getMessagePayload());
         }
     }
 
+
 }
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamHandler.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamHandler.java
new file mode 100644
index 0000000..4416e0a
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamHandler.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.netconf.ctl;
+
+import org.onosproject.netconf.NetconfDeviceOutputEventListener;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Interface to represent an objects that does all the IO on a NETCONF session
+ * with a device.
+ */
+public interface NetconfStreamHandler {
+    /**
+     * Sends the request on the stream that is used to communicate to and from the device.
+     *
+     * @param request request to send to the physical device
+     * @return a CompletableFuture of type String that will contain the response for the request.
+     */
+    CompletableFuture<String> sendMessage(String request);
+
+    /**
+     * Adds a listener for netconf events on the handled stream.
+     *
+     * @param listener Netconf device event listener
+     */
+    void addDeviceEventListener(NetconfDeviceOutputEventListener listener);
+
+    /**
+     * Removes a listener for netconf events on the handled stream.
+     *
+     * @param listener Netconf device event listener
+     */
+    void removeDeviceEventListener(NetconfDeviceOutputEventListener listener);
+}
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamThread.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamThread.java
new file mode 100644
index 0000000..6169c3d
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamThread.java
@@ -0,0 +1,225 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.netconf.ctl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfDeviceOutputEvent;
+import org.onosproject.netconf.NetconfDeviceOutputEventListener;
+import org.onosproject.netconf.NetconfException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Thread that gets spawned each time a session is established and handles all the input
+ * and output from the session's streams to and from the NETCONF device the session is
+ * established with.
+ */
+public class NetconfStreamThread extends Thread implements NetconfStreamHandler {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(NetconfStreamThread.class);
+    private static final String HELLO = "hello";
+    private static final String END_PATTERN = "]]>]]>";
+    private static final String RPC_REPLY = "rpc-reply";
+    private static final String RPC_ERROR = "rpc-error";
+    private static final String NOTIFICATION_LABEL = "<notification>";
+
+    private static PrintWriter outputStream;
+    private static NetconfDeviceInfo netconfDeviceInfo;
+    private static NetconfSessionDelegate sessionDelegate;
+    private static NetconfMessageState state;
+    private static List<NetconfDeviceOutputEventListener> netconfDeviceEventListeners
+            = Lists.newArrayList();
+
+    public NetconfStreamThread(final InputStream in, final OutputStream out,
+                               final InputStream err, NetconfDeviceInfo deviceInfo,
+                               NetconfSessionDelegate delegate) {
+        super(handler(in, err));
+        outputStream = new PrintWriter(out);
+        netconfDeviceInfo = deviceInfo;
+        state = NetconfMessageState.NO_MATCHING_PATTERN;
+        sessionDelegate = delegate;
+        log.debug("Stream thread for device {} session started", deviceInfo);
+        start();
+    }
+
+    @Override
+    public CompletableFuture<String> sendMessage(String request) {
+        outputStream.print(request);
+        outputStream.flush();
+        return new CompletableFuture<>();
+    }
+
+    public enum NetconfMessageState {
+
+        NO_MATCHING_PATTERN {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return FIRST_BRAKET;
+                } else {
+                    return this;
+                }
+            }
+        },
+        FIRST_BRAKET {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return SECOND_BRAKET;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        SECOND_BRAKET {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == '>') {
+                    return FIRST_BIGGER;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        FIRST_BIGGER {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return THIRD_BRAKET;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        THIRD_BRAKET {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return ENDING_BIGGER;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        ENDING_BIGGER {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == '>') {
+                    return END_PATTERN;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        END_PATTERN {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                return NO_MATCHING_PATTERN;
+            }
+        };
+
+        abstract NetconfMessageState evaluateChar(char c);
+    }
+
+    private static Runnable handler(final InputStream in, final InputStream err) {
+        BufferedReader bufferReader = new BufferedReader(new InputStreamReader(in));
+        return () -> {
+            try {
+                boolean socketClosed = false;
+                StringBuilder deviceReplyBuilder = new StringBuilder();
+                while (!socketClosed) {
+                    int cInt = bufferReader.read();
+                    if (cInt == -1) {
+                        socketClosed = true;
+                        NetconfDeviceOutputEvent event = new NetconfDeviceOutputEvent(
+                                NetconfDeviceOutputEvent.Type.DEVICE_UNREGISTERED,
+                                null, null, -1, netconfDeviceInfo);
+                        netconfDeviceEventListeners.forEach(
+                                listener -> listener.event(event));
+                    }
+                    char c = (char) cInt;
+                    state = state.evaluateChar(c);
+                    deviceReplyBuilder.append(c);
+                    if (state == NetconfMessageState.END_PATTERN) {
+                        String deviceReply = deviceReplyBuilder.toString()
+                                .replace(END_PATTERN, "");
+                        if (deviceReply.contains(RPC_REPLY) ||
+                                deviceReply.contains(RPC_ERROR) ||
+                                deviceReply.contains(HELLO)) {
+                            NetconfDeviceOutputEvent event = new NetconfDeviceOutputEvent(
+                                    NetconfDeviceOutputEvent.Type.DEVICE_REPLY,
+                                    null, deviceReply, getMsgId(deviceReply), netconfDeviceInfo);
+                            sessionDelegate.notify(event);
+                            netconfDeviceEventListeners.forEach(
+                                    listener -> listener.event(event));
+                        } else if (deviceReply.contains(NOTIFICATION_LABEL)) {
+                            final String finalDeviceReply = deviceReply;
+                            netconfDeviceEventListeners.forEach(
+                                    listener -> listener.event(new NetconfDeviceOutputEvent(
+                                            NetconfDeviceOutputEvent.Type.DEVICE_NOTIFICATION,
+                                            null, finalDeviceReply, getMsgId(finalDeviceReply), netconfDeviceInfo)));
+                        } else {
+                            log.info("Error on replay from device {} ", deviceReply);
+                        }
+                        deviceReplyBuilder.setLength(0);
+                    }
+                }
+            } catch (IOException e) {
+                log.warn("Error in reading from the session for device " + netconfDeviceInfo, e);
+                throw new RuntimeException(new NetconfException("Error in reading from the session for device {}" +
+                                                                        netconfDeviceInfo, e));
+                //TODO should we send a socket closed message to listeners ?
+            }
+        };
+    }
+
+    private static int getMsgId(String reply) {
+        if (!reply.contains(HELLO)) {
+            String[] outer = reply.split("message-id=");
+            Preconditions.checkArgument(outer.length != 1,
+                                        "Error in retrieving the message id");
+            String messageID = outer[1].substring(0, 3).replace("\"", "");
+            Preconditions.checkNotNull(Integer.parseInt(messageID),
+                                       "Error in retrieving the message id");
+            return Integer.parseInt(messageID);
+        } else {
+            return 0;
+        }
+    }
+
+    public void addDeviceEventListener(NetconfDeviceOutputEventListener listener) {
+        if (!netconfDeviceEventListeners.contains(listener)) {
+            netconfDeviceEventListeners.add(listener);
+        }
+    }
+
+    public void removeDeviceEventListener(NetconfDeviceOutputEventListener listener) {
+        netconfDeviceEventListeners.remove(listener);
+    }
+}