diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/DefaultNetconfDevice.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/DefaultNetconfDevice.java
new file mode 100644
index 0000000..0740cc4
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/DefaultNetconfDevice.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2016-present 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.NetconfDevice;
+import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfException;
+import org.onosproject.netconf.NetconfSession;
+import org.onosproject.netconf.NetconfSessionFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Default implementation of a NETCONF device.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+public class DefaultNetconfDevice implements NetconfDevice {
+
+    public static final Logger log = LoggerFactory
+            .getLogger(DefaultNetconfDevice.class);
+
+    private NetconfDeviceInfo netconfDeviceInfo;
+    private boolean deviceState = true;
+    protected NetconfSessionFactory sessionFactory = new SshNetconfSessionFactory();
+    private NetconfSession netconfSession;
+
+    /**
+     * Creates a new default NETCONF device with the information provided.
+     * The device gets created only if no exception is thrown while connecting to
+     * it and establishing the NETCONF session.
+     * @param deviceInfo information about the device to be created.
+     * @throws NetconfException if there are problems in creating or establishing
+     * the underlying NETCONF connection and session.
+     */
+    public DefaultNetconfDevice(NetconfDeviceInfo deviceInfo)
+            throws NetconfException {
+        netconfDeviceInfo = deviceInfo;
+        try {
+            netconfSession = sessionFactory.createNetconfSession(deviceInfo);
+        } catch (IOException e) {
+            deviceState = false;
+            throw new NetconfException("Cannot create connection and session for device " +
+                                               deviceInfo, e);
+        }
+    }
+
+    @Override
+    public boolean isActive() {
+        return deviceState;
+    }
+
+    @Override
+    public NetconfSession getSession() {
+        return netconfSession;
+    }
+
+    @Override
+    public void disconnect() {
+        deviceState = false;
+        try {
+            netconfSession.close();
+        } catch (IOException e) {
+            log.warn("Cannot communicate with the device {} session already closed", netconfDeviceInfo);
+        }
+    }
+
+    @Override
+    public NetconfDeviceInfo getDeviceInfo() {
+        return netconfDeviceInfo;
+    }
+
+    public class SshNetconfSessionFactory implements NetconfSessionFactory {
+
+        @Override
+        public NetconfSession createNetconfSession(NetconfDeviceInfo netconfDeviceInfo) throws NetconfException {
+            return new NetconfSessionImpl(netconfDeviceInfo);
+        }
+    }
+
+}
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
new file mode 100644
index 0000000..0747f8f
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfControllerImpl.java
@@ -0,0 +1,344 @@
+/*
+ * Copyright 2015-present 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.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Modified;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.IpAddress;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.key.DeviceKey;
+import org.onosproject.net.key.DeviceKeyId;
+import org.onosproject.net.key.DeviceKeyService;
+import org.onosproject.net.key.UsernamePassword;
+import org.onosproject.netconf.NetconfController;
+import org.onosproject.netconf.NetconfDevice;
+import org.onosproject.netconf.NetconfDeviceFactory;
+import org.onosproject.netconf.NetconfDeviceInfo;
+import org.onosproject.netconf.NetconfDeviceListener;
+import org.onosproject.netconf.NetconfDeviceOutputEvent;
+import org.onosproject.netconf.NetconfDeviceOutputEventListener;
+import org.onosproject.netconf.NetconfException;
+import org.osgi.service.component.ComponentContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.Dictionary;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static org.onlab.util.Tools.get;
+import static org.onlab.util.Tools.groupedThreads;
+
+/**
+ * The implementation of NetconfController.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+@Component(immediate = false, enabled = false)
+@Service
+public class NetconfControllerImpl implements NetconfController {
+    private static final int DEFAULT_CONNECT_TIMEOUT_SECONDS = 5;
+    private static final String PROP_NETCONF_CONNECT_TIMEOUT = "netconfConnectTimeout";
+    @Property(name = PROP_NETCONF_CONNECT_TIMEOUT, intValue = DEFAULT_CONNECT_TIMEOUT_SECONDS,
+            label = "Time (in seconds) to wait for a NETCONF connect.")
+    protected static int netconfConnectTimeout = DEFAULT_CONNECT_TIMEOUT_SECONDS;
+
+    private static final String PROP_NETCONF_REPLY_TIMEOUT = "netconfReplyTimeout";
+    private static final int DEFAULT_REPLY_TIMEOUT_SECONDS = 5;
+    @Property(name = PROP_NETCONF_REPLY_TIMEOUT, intValue = DEFAULT_REPLY_TIMEOUT_SECONDS,
+            label = "Time (in seconds) waiting for a NetConf reply")
+    protected static int netconfReplyTimeout = DEFAULT_REPLY_TIMEOUT_SECONDS;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ComponentConfigService cfgService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceKeyService deviceKeyService;
+
+    public static final Logger log = LoggerFactory
+            .getLogger(NetconfControllerImpl.class);
+
+    private Map<DeviceId, NetconfDevice> netconfDeviceMap = new ConcurrentHashMap<>();
+
+    private final NetconfDeviceOutputEventListener downListener = new DeviceDownEventListener();
+
+    protected Set<NetconfDeviceListener> netconfDeviceListeners = new CopyOnWriteArraySet<>();
+    protected NetconfDeviceFactory deviceFactory = new DefaultNetconfDeviceFactory();
+
+    protected final ExecutorService executor =
+            Executors.newCachedThreadPool(groupedThreads("onos/netconfdevicecontroller",
+                                                           "connection-reopen-%d", log));
+
+    @Activate
+    public void activate(ComponentContext context) {
+        cfgService.registerProperties(getClass());
+        modified(context);
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        cfgService.unregisterProperties(getClass(), false);
+        netconfDeviceMap.clear();
+        log.info("Stopped");
+    }
+
+    @Modified
+    public void modified(ComponentContext context) {
+        if (context == null) {
+            netconfReplyTimeout = DEFAULT_REPLY_TIMEOUT_SECONDS;
+            netconfConnectTimeout = DEFAULT_CONNECT_TIMEOUT_SECONDS;
+            log.info("No component configuration");
+            return;
+        }
+
+        Dictionary<?, ?> properties = context.getProperties();
+
+        int newNetconfReplyTimeout;
+        int newNetconfConnectTimeout;
+        try {
+            String s = get(properties, PROP_NETCONF_REPLY_TIMEOUT);
+            newNetconfReplyTimeout = isNullOrEmpty(s) ?
+                    netconfReplyTimeout : Integer.parseInt(s.trim());
+
+            s = get(properties, PROP_NETCONF_CONNECT_TIMEOUT);
+            newNetconfConnectTimeout = isNullOrEmpty(s) ?
+                    netconfConnectTimeout : Integer.parseInt(s.trim());
+
+        } catch (NumberFormatException e) {
+            log.warn("Component configuration had invalid value", e);
+            return;
+        }
+
+        if (newNetconfConnectTimeout < 0) {
+            log.warn("netconfConnectTimeout is invalid - less than 0");
+            return;
+        } else if (newNetconfReplyTimeout <= 0) {
+            log.warn("netconfReplyTimeout is invalid - 0 or less.");
+            return;
+        }
+
+        netconfReplyTimeout = newNetconfReplyTimeout;
+        netconfConnectTimeout = newNetconfConnectTimeout;
+        log.info("Settings: {} = {}, {} = {}",
+                 PROP_NETCONF_REPLY_TIMEOUT, netconfReplyTimeout, PROP_NETCONF_CONNECT_TIMEOUT, netconfConnectTimeout);
+    }
+
+    @Override
+    public void addDeviceListener(NetconfDeviceListener listener) {
+        if (!netconfDeviceListeners.contains(listener)) {
+            netconfDeviceListeners.add(listener);
+        }
+    }
+
+    @Override
+    public void removeDeviceListener(NetconfDeviceListener listener) {
+        netconfDeviceListeners.remove(listener);
+    }
+
+    @Override
+    public NetconfDevice getNetconfDevice(DeviceId deviceInfo) {
+        return netconfDeviceMap.get(deviceInfo);
+    }
+
+    @Override
+    public NetconfDevice getNetconfDevice(IpAddress ip, int port) {
+        for (DeviceId info : netconfDeviceMap.keySet()) {
+            if (info.uri().getSchemeSpecificPart().equals(ip.toString() + ":" + port)) {
+                return netconfDeviceMap.get(info);
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public NetconfDevice connectDevice(DeviceId deviceId) throws NetconfException {
+        if (netconfDeviceMap.containsKey(deviceId)) {
+            log.debug("Device {} is already present", deviceId);
+            return netconfDeviceMap.get(deviceId);
+        } else {
+            log.debug("Creating NETCONF device {}", deviceId);
+            Device device = deviceService.getDevice(deviceId);
+            String ip;
+            int port;
+            if (device != null) {
+                ip = device.annotations().value("ipaddress");
+                port = Integer.parseInt(device.annotations().value("port"));
+            } else {
+                String[] info = deviceId.toString().split(":");
+                if (info.length == 3) {
+                    ip = info[1];
+                    port = Integer.parseInt(info[2]);
+                } else {
+                    ip = Arrays.asList(info).stream().filter(el -> !el.equals(info[0])
+                            && !el.equals(info[info.length - 1]))
+                            .reduce((t, u) -> t + ":" + u)
+                            .get();
+                    log.debug("ip v6 {}", ip);
+                    port = Integer.parseInt(info[info.length - 1]);
+                }
+            }
+            try {
+                DeviceKey deviceKey = deviceKeyService.getDeviceKey(
+                        DeviceKeyId.deviceKeyId(deviceId.toString()));
+                NetconfDeviceInfo deviceInfo = null;
+                if (deviceKey.type() == DeviceKey.Type.USERNAME_PASSWORD) {
+                    UsernamePassword usernamepasswd = deviceKey.asUsernamePassword();
+
+                    deviceInfo = new NetconfDeviceInfo(usernamepasswd.username(),
+                                                       usernamepasswd.password(),
+                                                       IpAddress.valueOf(ip),
+                                                       port);
+
+                } else if (deviceKey.type() == DeviceKey.Type.SSL_KEY) {
+                    String username = deviceKey.annotations().value(AnnotationKeys.USERNAME);
+                    String password = deviceKey.annotations().value(AnnotationKeys.PASSWORD);
+                    String sshkey = deviceKey.annotations().value(AnnotationKeys.SSHKEY);
+
+                    deviceInfo = new NetconfDeviceInfo(username,
+                                                       password,
+                                                       IpAddress.valueOf(ip),
+                                                       port,
+                                                       sshkey);
+                } else {
+                    log.error("Unknown device key for device {}", deviceId);
+                }
+                NetconfDevice netconfDevicedevice = createDevice(deviceInfo);
+                netconfDevicedevice.getSession().addDeviceOutputListener(downListener);
+                return netconfDevicedevice;
+            } catch (NullPointerException e) {
+                throw new NetconfException("No Device Key for device " + deviceId, e);
+            }
+        }
+    }
+
+    @Override
+    public void disconnectDevice(DeviceId deviceId, boolean remove) {
+        if (!netconfDeviceMap.containsKey(deviceId)) {
+            log.warn("Device {} is not present", deviceId);
+        } else {
+            stopDevice(deviceId, remove);
+        }
+    }
+
+    private void stopDevice(DeviceId deviceId, boolean remove) {
+        netconfDeviceMap.get(deviceId).disconnect();
+        netconfDeviceMap.remove(deviceId);
+        if (remove) {
+            for (NetconfDeviceListener l : netconfDeviceListeners) {
+                l.deviceRemoved(deviceId);
+            }
+        }
+    }
+
+    @Override
+    public void removeDevice(DeviceId deviceId) {
+        if (!netconfDeviceMap.containsKey(deviceId)) {
+            log.warn("Device {} is not present", deviceId);
+            for (NetconfDeviceListener l : netconfDeviceListeners) {
+                l.deviceRemoved(deviceId);
+            }
+        } else {
+            netconfDeviceMap.remove(deviceId);
+            for (NetconfDeviceListener l : netconfDeviceListeners) {
+                l.deviceRemoved(deviceId);
+            }
+        }
+    }
+
+    private NetconfDevice createDevice(NetconfDeviceInfo deviceInfo) throws NetconfException {
+        NetconfDevice netconfDevice = deviceFactory.createNetconfDevice(deviceInfo);
+        netconfDeviceMap.put(deviceInfo.getDeviceId(), netconfDevice);
+        for (NetconfDeviceListener l : netconfDeviceListeners) {
+            l.deviceAdded(deviceInfo.getDeviceId());
+        }
+        return netconfDevice;
+    }
+
+
+    @Override
+    public Map<DeviceId, NetconfDevice> getDevicesMap() {
+        return netconfDeviceMap;
+    }
+
+    @Override
+    public Set<DeviceId> getNetconfDevices() {
+        return netconfDeviceMap.keySet();
+    }
+
+
+    //Device factory for the specific NetconfDeviceImpl
+    private class DefaultNetconfDeviceFactory implements NetconfDeviceFactory {
+
+        @Override
+        public NetconfDevice createNetconfDevice(NetconfDeviceInfo netconfDeviceInfo) throws NetconfException {
+            return new DefaultNetconfDevice(netconfDeviceInfo);
+        }
+    }
+
+    //Listener for closed session with devices, gets triggered whe devices goes down
+    // or sends the endpattern ]]>]]>
+    private class DeviceDownEventListener implements NetconfDeviceOutputEventListener {
+
+        @Override
+        public void event(NetconfDeviceOutputEvent event) {
+            DeviceId did = event.getDeviceInfo().getDeviceId();
+            if (event.type().equals(NetconfDeviceOutputEvent.Type.DEVICE_UNREGISTERED)) {
+                removeDevice(did);
+            } else if (event.type().equals(NetconfDeviceOutputEvent.Type.SESSION_CLOSED)) {
+                log.info("Trying to reestablish connection with device {}", did);
+                executor.execute(() -> {
+                    try {
+                        netconfDeviceMap.get(did).getSession().checkAndReestablish();
+                        log.info("Connection with device {} was reestablished", did);
+                    } catch (NetconfException e) {
+                        log.error("The SSH connection with device {} couldn't be " +
+                                          "reestablished due to {}. " +
+                                          "Marking the device as unreachable", e.getMessage());
+                        log.debug("Complete exception: ", e);
+                        removeDevice(did);
+                    }
+                });
+            }
+        }
+
+        @Override
+        public boolean isRelevant(NetconfDeviceOutputEvent event) {
+            return getDevicesMap().containsKey(event.getDeviceInfo().getDeviceId());
+        }
+    }
+}
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..305b785
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionDelegate.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015-present 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.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+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
new file mode 100644
index 0000000..00bbca0
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfSessionImpl.java
@@ -0,0 +1,670 @@
+/*
+ * Copyright 2015-present 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.annotations.Beta;
+import ch.ethz.ssh2.Connection;
+import ch.ethz.ssh2.Session;
+import ch.ethz.ssh2.channel.Channel;
+import com.google.common.base.Preconditions;
+import org.onosproject.netconf.TargetConfig;
+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.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Pattern;
+
+
+/**
+ * Implementation of a NETCONF session to talk to a device.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+public class NetconfSessionImpl implements NetconfSession {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(NetconfSessionImpl.class);
+
+    private static final String ENDPATTERN = "]]>]]>";
+    private static final String MESSAGE_ID_STRING = "message-id";
+    private static final String HELLO = "<hello";
+    private static final String NEW_LINE = "\n";
+    private static final String END_OF_RPC_OPEN_TAG = "\">";
+    private static final String EQUAL = "=";
+    private static final String NUMBER_BETWEEN_QUOTES_MATCHER = "\"+([0-9]+)+\"";
+    private static final String RPC_OPEN = "<rpc ";
+    private static final String RPC_CLOSE = "</rpc>";
+    private static final String GET_OPEN = "<get>";
+    private static final String GET_CLOSE = "</get>";
+    private static final String WITH_DEFAULT_OPEN = "<with-defaults ";
+    private static final String WITH_DEFAULT_CLOSE = "</with-defaults>";
+    private static final String DEFAULT_OPERATION_OPEN = "<default-operation>";
+    private static final String DEFAULT_OPERATION_CLOSE = "</default-operation>";
+    private static final String SUBTREE_FILTER_OPEN = "<filter type=\"subtree\">";
+    private static final String SUBTREE_FILTER_CLOSE = "</filter>";
+    private static final String EDIT_CONFIG_OPEN = "<edit-config>";
+    private static final String EDIT_CONFIG_CLOSE = "</edit-config>";
+    private static final String TARGET_OPEN = "<target>";
+    private static final String TARGET_CLOSE = "</target>";
+    private static final String CONFIG_OPEN = "<config xmlns:nc=\"urn:ietf:params:xml:ns:netconf:base:1.0\">";
+    private static final String CONFIG_CLOSE = "</config>";
+    private static final String XML_HEADER =
+            "<?xml version=\"1.0\" encoding=\"UTF-8\"?>";
+    private static final String NETCONF_BASE_NAMESPACE =
+            "xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\"";
+    private static final String NETCONF_WITH_DEFAULTS_NAMESPACE =
+            "xmlns=\"urn:ietf:params:xml:ns:yang:ietf-netconf-with-defaults\"";
+    private static final String SUBSCRIPTION_SUBTREE_FILTER_OPEN =
+            "<filter xmlns:base10=\"urn:ietf:params:xml:ns:netconf:base:1.0\" base10:type=\"subtree\">";
+
+    private static Pattern msgIdPattern = Pattern.compile("(message-id=\"[0-9]+\")");
+
+    private final AtomicInteger messageIdInteger = new AtomicInteger(0);
+    private Connection netconfConnection;
+    private NetconfDeviceInfo deviceInfo;
+    private Session sshSession;
+    private boolean connectionActive;
+    private List<String> deviceCapabilities =
+            Collections.singletonList("urn:ietf:params:netconf:base:1.0");
+    private String serverCapabilities;
+    private NetconfStreamHandler streamHandler;
+    private Map<Integer, CompletableFuture<String>> replies;
+    private List<String> errorReplies;
+    private boolean subscriptionConnected = false;
+    private String notificationFilterSchema = null;
+
+
+    public NetconfSessionImpl(NetconfDeviceInfo deviceInfo) throws NetconfException {
+        this.deviceInfo = deviceInfo;
+        this.netconfConnection = null;
+        this.sshSession = null;
+        connectionActive = false;
+        replies = new ConcurrentHashMap<>();
+        errorReplies = new ArrayList<>();
+        startConnection();
+    }
+
+    private void startConnection() throws NetconfException {
+        if (!connectionActive) {
+            netconfConnection = new Connection(deviceInfo.ip().toString(), deviceInfo.port());
+            int connectTimeout = NetconfControllerImpl.netconfConnectTimeout;
+
+            try {
+                netconfConnection.connect(null, 1000 * connectTimeout, 1000 * connectTimeout);
+            } catch (IOException e) {
+                throw new NetconfException("Cannot open a connection with device " + deviceInfo, e);
+            }
+            boolean isAuthenticated;
+            try {
+                if (deviceInfo.getKeyFile() != null && deviceInfo.getKeyFile().canRead()) {
+                    log.debug("Authenticating with key file to device {} with username {}",
+                              deviceInfo.getDeviceId(), deviceInfo.name());
+                    isAuthenticated = netconfConnection.authenticateWithPublicKey(
+                            deviceInfo.name(), deviceInfo.getKeyFile(),
+                            deviceInfo.password().equals("") ? null : deviceInfo.password());
+                } else if (deviceInfo.getKey() != null) {
+                    log.debug("Authenticating with key to device {} with username {}",
+                              deviceInfo.getDeviceId(), deviceInfo.name());
+                    isAuthenticated = netconfConnection.authenticateWithPublicKey(
+                            deviceInfo.name(), deviceInfo.getKey(),
+                            deviceInfo.password().equals("") ? null : deviceInfo.password());
+                } else {
+                    log.debug("Authenticating to device {} with username {} with password",
+                              deviceInfo.getDeviceId(), deviceInfo.name());
+                    isAuthenticated = netconfConnection.authenticateWithPassword(
+                            deviceInfo.name(), deviceInfo.password());
+                }
+            } catch (IOException e) {
+                log.error("Authentication connection to device {} failed",
+                          deviceInfo.getDeviceId(), e);
+                throw new NetconfException("Authentication connection to device " +
+                                                   deviceInfo.getDeviceId() + " failed", e);
+            }
+
+            connectionActive = true;
+            Preconditions.checkArgument(isAuthenticated,
+                                        "Authentication to device %s with username " +
+                                                "%s failed",
+                                        deviceInfo.getDeviceId(), deviceInfo.name());
+            startSshSession();
+        }
+    }
+
+    private void startSshSession() throws NetconfException {
+        try {
+            sshSession = netconfConnection.openSession();
+            sshSession.startSubSystem("netconf");
+            streamHandler = new NetconfStreamThread(sshSession.getStdout(), sshSession.getStdin(),
+                                                    sshSession.getStderr(), deviceInfo,
+                                                    new NetconfSessionDelegateImpl(),
+                                                    replies);
+            this.addDeviceOutputListener(new NetconfDeviceOutputEventListenerImpl(deviceInfo));
+            sendHello();
+        } catch (IOException e) {
+            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);
+        }
+    }
+
+
+    @Beta
+    private void startSubscriptionConnection(String filterSchema) throws NetconfException {
+        if (!serverCapabilities.contains("interleave")) {
+            throw new NetconfException("Device" + deviceInfo + "does not support interleave");
+        }
+        String reply = sendRequest(createSubscriptionString(filterSchema));
+        if (!checkReply(reply)) {
+            throw new NetconfException("Subscription not successful with device "
+                                               + deviceInfo + " with reply " + reply);
+        }
+        subscriptionConnected = true;
+    }
+
+    @Override
+    public void startSubscription() throws NetconfException {
+        if (!subscriptionConnected) {
+            startSubscriptionConnection(null);
+        }
+        streamHandler.setEnableNotifications(true);
+    }
+
+    @Beta
+    @Override
+    public void startSubscription(String filterSchema) throws NetconfException {
+        if (!subscriptionConnected) {
+            notificationFilterSchema = filterSchema;
+            startSubscriptionConnection(filterSchema);
+        }
+        streamHandler.setEnableNotifications(true);
+    }
+
+    @Beta
+    private String createSubscriptionString(String filterSchema) {
+        StringBuilder subscriptionbuffer = new StringBuilder();
+        subscriptionbuffer.append("<rpc xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
+        subscriptionbuffer.append("  <create-subscription\n");
+        subscriptionbuffer.append("xmlns=\"urn:ietf:params:xml:ns:netconf:notification:1.0\">\n");
+        // FIXME Only subtree filtering supported at the moment.
+        if (filterSchema != null) {
+            subscriptionbuffer.append("    ");
+            subscriptionbuffer.append(SUBSCRIPTION_SUBTREE_FILTER_OPEN).append(NEW_LINE);
+            subscriptionbuffer.append(filterSchema).append(NEW_LINE);
+            subscriptionbuffer.append("    ");
+            subscriptionbuffer.append(SUBTREE_FILTER_CLOSE).append(NEW_LINE);
+        }
+        subscriptionbuffer.append("  </create-subscription>\n");
+        subscriptionbuffer.append("</rpc>\n");
+        subscriptionbuffer.append(ENDPATTERN);
+        return subscriptionbuffer.toString();
+    }
+
+    @Override
+    public void endSubscription() throws NetconfException {
+        if (subscriptionConnected) {
+            streamHandler.setEnableNotifications(false);
+        } else {
+            throw new NetconfException("Subscription does not exist.");
+        }
+    }
+
+    private void sendHello() throws NetconfException {
+        serverCapabilities = sendRequest(createHelloString());
+    }
+
+    private String createHelloString() {
+        StringBuilder hellobuffer = new StringBuilder();
+        hellobuffer.append(XML_HEADER);
+        hellobuffer.append("\n");
+        hellobuffer.append("<hello xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
+        hellobuffer.append("  <capabilities>\n");
+        deviceCapabilities.forEach(
+                cap -> hellobuffer.append("    <capability>")
+                        .append(cap)
+                        .append("</capability>\n"));
+        hellobuffer.append("  </capabilities>\n");
+        hellobuffer.append("</hello>\n");
+        hellobuffer.append(ENDPATTERN);
+        return hellobuffer.toString();
+
+    }
+
+    @Override
+    public void checkAndReestablish() throws NetconfException {
+        if (sshSession.getState() != Channel.STATE_OPEN) {
+            try {
+                log.debug("Trying to reopen the Sesion with {}", deviceInfo.getDeviceId());
+                startSshSession();
+            } catch (IOException | IllegalStateException e) {
+                log.debug("Trying to reopen the Connection with {}", deviceInfo.getDeviceId());
+                try {
+                    connectionActive = false;
+                    replies.clear();
+                    messageIdInteger.set(0);
+                    startConnection();
+                    if (subscriptionConnected) {
+                        log.debug("Restarting subscription with {}", deviceInfo.getDeviceId());
+                        subscriptionConnected = false;
+                        startSubscription(notificationFilterSchema);
+                    }
+                } catch (IOException e2) {
+                    log.error("No connection {} for device {}", netconfConnection, e.getMessage());
+                    throw new NetconfException("Cannot re-open the connection with device" + deviceInfo, e);
+                }
+            }
+        }
+    }
+
+    @Override
+    public String requestSync(String request) throws NetconfException {
+        if (!request.contains(ENDPATTERN)) {
+            request = request + NEW_LINE + ENDPATTERN;
+        }
+        String reply = sendRequest(request);
+        checkReply(reply);
+        return reply;
+    }
+
+    @Override
+    @Deprecated
+    public CompletableFuture<String> request(String request) {
+        return streamHandler.sendMessage(request);
+    }
+
+    private CompletableFuture<String> request(String request, int messageId) {
+        return streamHandler.sendMessage(request, messageId);
+    }
+
+    private String sendRequest(String request) throws NetconfException {
+        checkAndReestablish();
+        final int messageId = messageIdInteger.getAndIncrement();
+        request = formatRequestMessageId(request, messageId);
+        request = formatXmlHeader(request);
+        CompletableFuture<String> futureReply = request(request, messageId);
+        int replyTimeout = NetconfControllerImpl.netconfReplyTimeout;
+        String rp;
+        try {
+            rp = futureReply.get(replyTimeout, TimeUnit.SECONDS);
+            replies.remove(messageId);
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            throw new NetconfException("No matching reply for request " + request, e);
+        }
+        log.debug("Result {} from request {} to device {}", rp, request, deviceInfo);
+        return rp.trim();
+    }
+
+    private String formatRequestMessageId(String request, int messageId) {
+        if (request.contains(MESSAGE_ID_STRING)) {
+            //FIXME if application provides his own counting of messages this fails that count
+            request = request.replaceFirst(MESSAGE_ID_STRING + EQUAL + NUMBER_BETWEEN_QUOTES_MATCHER,
+                                           MESSAGE_ID_STRING + EQUAL + "\"" + messageId + "\"");
+        } else if (!request.contains(MESSAGE_ID_STRING) && !request.contains(HELLO)) {
+            //FIXME find out a better way to enforce the presence of message-id
+            request = request.replaceFirst(END_OF_RPC_OPEN_TAG, "\" " + MESSAGE_ID_STRING + EQUAL + "\""
+                    + messageId + "\"" + ">");
+        }
+        return request;
+    }
+
+    private String formatXmlHeader(String request) {
+        if (!request.contains(XML_HEADER)) {
+            //FIXME if application provieds his own XML header of different type there is a clash
+            request = XML_HEADER + "\n" + request;
+        }
+        return request;
+    }
+
+    @Override
+    public String doWrappedRpc(String request) throws NetconfException {
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append(RPC_OPEN);
+        rpc.append(MESSAGE_ID_STRING);
+        rpc.append(EQUAL);
+        rpc.append("\"");
+        rpc.append(messageIdInteger.get());
+        rpc.append("\"  ");
+        rpc.append(NETCONF_BASE_NAMESPACE).append(">\n");
+        rpc.append(request);
+        rpc.append(RPC_CLOSE).append(NEW_LINE);
+        rpc.append(ENDPATTERN);
+        String reply = sendRequest(rpc.toString());
+        checkReply(reply);
+        return reply;
+    }
+
+    @Override
+    public String get(String request) throws NetconfException {
+        return requestSync(request);
+    }
+
+    @Override
+    public String get(String filterSchema, String withDefaultsMode) throws NetconfException {
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append(RPC_OPEN);
+        rpc.append(MESSAGE_ID_STRING);
+        rpc.append(EQUAL);
+        rpc.append("\"");
+        rpc.append(messageIdInteger.get());
+        rpc.append("\"  ");
+        rpc.append(NETCONF_BASE_NAMESPACE).append(">\n");
+        rpc.append(GET_OPEN).append(NEW_LINE);
+        if (filterSchema != null) {
+            rpc.append(SUBTREE_FILTER_OPEN).append(NEW_LINE);
+            rpc.append(filterSchema).append(NEW_LINE);
+            rpc.append(SUBTREE_FILTER_CLOSE).append(NEW_LINE);
+        }
+        if (withDefaultsMode != null) {
+            rpc.append(WITH_DEFAULT_OPEN).append(NETCONF_WITH_DEFAULTS_NAMESPACE).append(">");
+            rpc.append(withDefaultsMode).append(WITH_DEFAULT_CLOSE).append(NEW_LINE);
+        }
+        rpc.append(GET_CLOSE).append(NEW_LINE);
+        rpc.append(RPC_CLOSE).append(NEW_LINE);
+        rpc.append(ENDPATTERN);
+        String reply = sendRequest(rpc.toString());
+        checkReply(reply);
+        return reply;
+    }
+
+    @Override
+    public String getConfig(TargetConfig netconfTargetConfig) throws NetconfException {
+        return getConfig(netconfTargetConfig, null);
+    }
+
+    @Override
+    public String getConfig(String netconfTargetConfig) throws NetconfException {
+        return getConfig(TargetConfig.toTargetConfig(netconfTargetConfig));
+    }
+
+    @Override
+    public String getConfig(String netconfTargetConfig, String configurationFilterSchema) throws NetconfException {
+        return getConfig(TargetConfig.toTargetConfig(netconfTargetConfig), configurationFilterSchema);
+    }
+
+    @Override
+    public String getConfig(TargetConfig netconfTargetConfig, String configurationSchema) throws NetconfException {
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append("<rpc ");
+        rpc.append(MESSAGE_ID_STRING);
+        rpc.append(EQUAL);
+        rpc.append("\"");
+        rpc.append(messageIdInteger.get());
+        rpc.append("\"  ");
+        rpc.append("xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
+        rpc.append("<get-config>\n");
+        rpc.append("<source>\n");
+        rpc.append("<").append(netconfTargetConfig).append("/>");
+        rpc.append("</source>");
+        if (configurationSchema != null) {
+            rpc.append("<filter type=\"subtree\">\n");
+            rpc.append(configurationSchema).append("\n");
+            rpc.append("</filter>\n");
+        }
+        rpc.append("</get-config>\n");
+        rpc.append("</rpc>\n");
+        rpc.append(ENDPATTERN);
+        String reply = sendRequest(rpc.toString());
+        return checkReply(reply) ? reply : "ERROR " + reply;
+    }
+
+    @Override
+    public boolean editConfig(String newConfiguration) throws NetconfException {
+        newConfiguration = newConfiguration + ENDPATTERN;
+        return checkReply(sendRequest(newConfiguration));
+    }
+
+    @Override
+    public boolean editConfig(String netconfTargetConfig, String mode, String newConfiguration)
+            throws NetconfException {
+        return editConfig(TargetConfig.toTargetConfig(netconfTargetConfig), mode, newConfiguration);
+    }
+
+    @Override
+    public boolean editConfig(TargetConfig netconfTargetConfig, String mode, String newConfiguration)
+            throws NetconfException {
+        newConfiguration = newConfiguration.trim();
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append(RPC_OPEN);
+        rpc.append(MESSAGE_ID_STRING);
+        rpc.append(EQUAL);
+        rpc.append("\"");
+        rpc.append(messageIdInteger.get());
+        rpc.append("\"  ");
+        rpc.append(NETCONF_BASE_NAMESPACE).append(">\n");
+        rpc.append(EDIT_CONFIG_OPEN).append("\n");
+        rpc.append(TARGET_OPEN);
+        rpc.append("<").append(netconfTargetConfig).append("/>");
+        rpc.append(TARGET_CLOSE).append("\n");
+        if (mode != null) {
+            rpc.append(DEFAULT_OPERATION_OPEN);
+            rpc.append(mode);
+            rpc.append(DEFAULT_OPERATION_CLOSE).append("\n");
+        }
+        rpc.append(CONFIG_OPEN).append("\n");
+        rpc.append(newConfiguration);
+        rpc.append(CONFIG_CLOSE).append("\n");
+        rpc.append(EDIT_CONFIG_CLOSE).append("\n");
+        rpc.append(RPC_CLOSE);
+        rpc.append(ENDPATTERN);
+        log.debug(rpc.toString());
+        String reply = sendRequest(rpc.toString());
+        return checkReply(reply);
+    }
+
+    @Override
+    public boolean copyConfig(String netconfTargetConfig, String newConfiguration) throws NetconfException {
+        return copyConfig(TargetConfig.toTargetConfig(netconfTargetConfig), newConfiguration);
+    }
+
+    @Override
+    public boolean copyConfig(TargetConfig netconfTargetConfig, String newConfiguration)
+            throws NetconfException {
+        newConfiguration = newConfiguration.trim();
+        if (!newConfiguration.startsWith("<config>")) {
+            newConfiguration = "<config>" + newConfiguration
+                    + "</config>";
+        }
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append(RPC_OPEN);
+        rpc.append(NETCONF_BASE_NAMESPACE).append(">\n");
+        rpc.append("<copy-config>");
+        rpc.append("<target>");
+        rpc.append("<").append(netconfTargetConfig).append("/>");
+        rpc.append("</target>");
+        rpc.append("<source>");
+        rpc.append(newConfiguration);
+        rpc.append("</source>");
+        rpc.append("</copy-config>");
+        rpc.append("</rpc>");
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
+    }
+
+    @Override
+    public boolean deleteConfig(String netconfTargetConfig) throws NetconfException {
+        return deleteConfig(TargetConfig.toTargetConfig(netconfTargetConfig));
+    }
+
+    @Override
+    public boolean deleteConfig(TargetConfig netconfTargetConfig) throws NetconfException {
+        if (netconfTargetConfig.equals(TargetConfig.RUNNING)) {
+            log.warn("Target configuration for delete operation can't be \"running\"",
+                     netconfTargetConfig);
+            return false;
+        }
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append("<rpc>");
+        rpc.append("<delete-config>");
+        rpc.append("<target>");
+        rpc.append("<").append(netconfTargetConfig).append("/>");
+        rpc.append("</target>");
+        rpc.append("</delete-config>");
+        rpc.append("</rpc>");
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString()));
+    }
+
+    @Override
+    public boolean lock(String configType) throws NetconfException {
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append("<rpc xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
+        rpc.append("<lock>");
+        rpc.append("<target>");
+        rpc.append("<");
+        rpc.append(configType);
+        rpc.append("/>");
+        rpc.append("</target>");
+        rpc.append("</lock>");
+        rpc.append("</rpc>");
+        rpc.append(ENDPATTERN);
+        String lockReply = sendRequest(rpc.toString());
+        return checkReply(lockReply);
+    }
+
+    @Override
+    public boolean unlock(String configType) throws NetconfException {
+        StringBuilder rpc = new StringBuilder(XML_HEADER);
+        rpc.append("<rpc xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">\n");
+        rpc.append("<unlock>");
+        rpc.append("<target>");
+        rpc.append("<");
+        rpc.append(configType);
+        rpc.append("/>");
+        rpc.append("</target>");
+        rpc.append("</unlock>");
+        rpc.append("</rpc>");
+        rpc.append(ENDPATTERN);
+        String unlockReply = sendRequest(rpc.toString());
+        return checkReply(unlockReply);
+    }
+
+    @Override
+    public boolean lock() throws NetconfException {
+        return lock("running");
+    }
+
+    @Override
+    public boolean unlock() throws NetconfException {
+        return unlock("running");
+    }
+
+    @Override
+    public boolean close() throws NetconfException {
+        return close(false);
+    }
+
+    private boolean close(boolean force) throws NetconfException {
+        StringBuilder rpc = new StringBuilder();
+        rpc.append("<rpc xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">");
+        if (force) {
+            rpc.append("<kill-session/>");
+        } else {
+            rpc.append("<close-session/>");
+        }
+        rpc.append("</rpc>");
+        rpc.append(ENDPATTERN);
+        return checkReply(sendRequest(rpc.toString())) || close(true);
+    }
+
+    @Override
+    public String getSessionId() {
+        if (serverCapabilities.contains("<session-id>")) {
+            String[] outer = serverCapabilities.split("<session-id>");
+            Preconditions.checkArgument(outer.length != 1,
+                                        "Error in retrieving the session id");
+            String[] value = outer[1].split("</session-id>");
+            Preconditions.checkArgument(value.length != 1,
+                                        "Error in retrieving the session id");
+            return value[0];
+        } else {
+            return String.valueOf(-1);
+        }
+    }
+
+    @Override
+    public String getServerCapabilities() {
+        return serverCapabilities;
+    }
+
+    @Override
+    public void setDeviceCapabilities(List<String> capabilities) {
+        deviceCapabilities = capabilities;
+    }
+
+    @Override
+    public void addDeviceOutputListener(NetconfDeviceOutputEventListener listener) {
+        streamHandler.addDeviceEventListener(listener);
+    }
+
+    @Override
+    public void removeDeviceOutputListener(NetconfDeviceOutputEventListener listener) {
+        streamHandler.removeDeviceEventListener(listener);
+    }
+
+    private boolean checkReply(String reply) throws NetconfException {
+        if (reply != null) {
+            if (!reply.contains("<rpc-error>")) {
+                log.debug("Device {} sent reply {}", deviceInfo, reply);
+                return true;
+            } else if (reply.contains("<ok/>")
+                    || (reply.contains("<rpc-error>")
+                    && reply.contains("warning"))) {
+                log.debug("Device {} sent reply {}", deviceInfo, reply);
+                return true;
+            }
+        }
+        log.warn("Device {} has error in reply {}", deviceInfo, reply);
+        return false;
+    }
+
+    public class NetconfSessionDelegateImpl implements NetconfSessionDelegate {
+
+        @Override
+        public void notify(NetconfDeviceOutputEvent event) {
+            Optional<Integer> messageId = event.getMessageID();
+            log.debug("messageID {}, waiting replies messageIDs {}", messageId,
+                      replies.keySet());
+            if (!messageId.isPresent()) {
+                errorReplies.add(event.getMessagePayload());
+                log.error("Device {} sent error reply {}",
+                          event.getDeviceInfo(), event.getMessagePayload());
+                return;
+            }
+            CompletableFuture<String> completedReply =
+                    replies.get(messageId.get());
+            if (completedReply != null) {
+                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..2f39711
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamHandler.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2015-present 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.annotations.Beta;
+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.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+public interface NetconfStreamHandler {
+    /**
+     * Sends the request on the stream that is used to communicate to and from the device.
+     *
+     * If this request does not contain a messageId then this will throw a NoSuchElementException
+     *
+     * @param request request to send to the physical device
+     * @return a CompletableFuture of type String that will contain the response for the request.
+     * @deprecated - use method with messageId parameter instead
+     */
+    @Deprecated
+    CompletableFuture<String> sendMessage(String request);
+
+    /**
+     * 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
+     * @param messageId The identifier of the message - should be unique for the session
+     * @return a CompletableFuture of type String that will contain the response for the request.
+     */
+    CompletableFuture<String> sendMessage(String request, int messageId);
+
+    /**
+     * 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);
+
+    @Beta
+    /**
+     * Sets instance variable that when true allows receipt of notifications.
+     *
+     * @param enableNotifications if true, allows action based off notifications
+     *                             else, stops action based off notifications
+     */
+    void setEnableNotifications(boolean enableNotifications);
+}
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..ec6346a
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/NetconfStreamThread.java
@@ -0,0 +1,283 @@
+/*
+ * Copyright 2015-present 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.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * 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.
+ *
+ * @deprecated in 1.10.0
+ */
+@Deprecated
+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 final String MESSAGE_ID = "message-id=";
+    private static final Pattern MSGID_PATTERN = Pattern.compile(MESSAGE_ID + "\"(\\d+)\"");
+
+    private PrintWriter outputStream;
+    private final InputStream err;
+    private final InputStream in;
+    private NetconfDeviceInfo netconfDeviceInfo;
+    private NetconfSessionDelegate sessionDelegate;
+    private NetconfMessageState state;
+    private List<NetconfDeviceOutputEventListener> netconfDeviceEventListeners
+            = Lists.newCopyOnWriteArrayList();
+    private boolean enableNotifications = true;
+    private Map<Integer, CompletableFuture<String>> replies;
+
+    public NetconfStreamThread(final InputStream in, final OutputStream out,
+                               final InputStream err, NetconfDeviceInfo deviceInfo,
+                               NetconfSessionDelegate delegate,
+                               Map<Integer, CompletableFuture<String>> replies) {
+        this.in = in;
+        this.err = err;
+        outputStream = new PrintWriter(out);
+        netconfDeviceInfo = deviceInfo;
+        state = NetconfMessageState.NO_MATCHING_PATTERN;
+        sessionDelegate = delegate;
+        this.replies = replies;
+        log.debug("Stream thread for device {} session started", deviceInfo);
+        start();
+    }
+
+    @Override
+    public CompletableFuture<String> sendMessage(String request) {
+        Optional<Integer> messageId = getMsgId(request);
+        return sendMessage(request, messageId.get());
+    }
+
+    @Override
+    public CompletableFuture<String> sendMessage(String request, int messageId) {
+        log.debug("Sending message {} to device {}", request, netconfDeviceInfo);
+        CompletableFuture<String> cf = new CompletableFuture<>();
+        replies.put(messageId, cf);
+
+        synchronized (outputStream) {
+            outputStream.print(request);
+            outputStream.flush();
+        }
+
+        return cf;
+    }
+
+    public enum NetconfMessageState {
+
+        NO_MATCHING_PATTERN {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return FIRST_BRACKET;
+                } else {
+                    return this;
+                }
+            }
+        },
+        FIRST_BRACKET {
+            @Override
+            NetconfMessageState evaluateChar(char c) {
+                if (c == ']') {
+                    return SECOND_BRACKET;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        SECOND_BRACKET {
+            @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_BRACKET;
+                } else {
+                    return NO_MATCHING_PATTERN;
+                }
+            }
+        },
+        THIRD_BRACKET {
+            @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);
+    }
+
+    @Override
+    public void run() {
+        BufferedReader bufferReader = new BufferedReader(new InputStreamReader(in));
+            try {
+                boolean socketClosed = false;
+                StringBuilder deviceReplyBuilder = new StringBuilder();
+                while (!socketClosed) {
+                    int cInt = bufferReader.read();
+                    if (cInt == -1) {
+                        log.debug("Netconf device {}  sent error char in session," +
+                                          " will need to be reopend", netconfDeviceInfo);
+                        NetconfDeviceOutputEvent event = new NetconfDeviceOutputEvent(
+                                NetconfDeviceOutputEvent.Type.SESSION_CLOSED,
+                                null, null, Optional.of(-1), netconfDeviceInfo);
+                        netconfDeviceEventListeners.forEach(
+                                listener -> listener.event(event));
+                        socketClosed = true;
+                        log.debug("Netconf device {} ERROR cInt == -1 socketClosed = true", netconfDeviceInfo);
+                    }
+                    char c = (char) cInt;
+                    state = state.evaluateChar(c);
+                    deviceReplyBuilder.append(c);
+                    if (state == NetconfMessageState.END_PATTERN) {
+                        String deviceReply = deviceReplyBuilder.toString();
+                        if (deviceReply.equals(END_PATTERN)) {
+                            socketClosed = true;
+                            log.debug("Netconf device {} socketClosed = true DEVICE_UNREGISTERED {}",
+                                     netconfDeviceInfo, deviceReply);
+                            NetconfDeviceOutputEvent event = new NetconfDeviceOutputEvent(
+                                    NetconfDeviceOutputEvent.Type.DEVICE_UNREGISTERED,
+                                    null, null, Optional.of(-1), netconfDeviceInfo);
+                            netconfDeviceEventListeners.forEach(
+                                    listener -> listener.event(event));
+                            this.interrupt();
+                        } else {
+                            deviceReply = deviceReply.replace(END_PATTERN, "");
+                            if (deviceReply.contains(RPC_REPLY) ||
+                                    deviceReply.contains(RPC_ERROR) ||
+                                    deviceReply.contains(HELLO)) {
+                                log.debug("Netconf device {} sessionDelegate.notify() DEVICE_REPLY {} {}",
+                                    netconfDeviceInfo, getMsgId(deviceReply), deviceReply);
+                                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)) {
+                                log.debug("Netconf device {} DEVICE_NOTIFICATION {} {} {}",
+                                         netconfDeviceInfo, enableNotifications,
+                                         getMsgId(deviceReply), deviceReply);
+                                if (enableNotifications) {
+                                    log.debug("dispatching to {} listeners", netconfDeviceEventListeners.size());
+                                    final String finalDeviceReply = deviceReply;
+                                    netconfDeviceEventListeners.forEach(
+                                            listener -> listener.event(new NetconfDeviceOutputEvent(
+                                                    NetconfDeviceOutputEvent.Type.DEVICE_NOTIFICATION,
+                                                    null, finalDeviceReply, getMsgId(finalDeviceReply),
+                                                    netconfDeviceInfo)));
+                                }
+                            } else {
+                                log.debug("Error on reply from device {} {}", netconfDeviceInfo, 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 ?
+            }
+    }
+
+    protected static Optional<Integer> getMsgId(String reply) {
+        Matcher matcher = MSGID_PATTERN.matcher(reply);
+        if (matcher.find()) {
+            Integer messageId = Integer.parseInt(matcher.group(1));
+            Preconditions.checkNotNull(messageId, "Error in retrieving the message id");
+            return Optional.of(messageId);
+        }
+        if (reply.contains(HELLO)) {
+            return Optional.of(0);
+        }
+        return Optional.empty();
+    }
+
+    @Override
+    public void addDeviceEventListener(NetconfDeviceOutputEventListener listener) {
+        if (!netconfDeviceEventListeners.contains(listener)) {
+            netconfDeviceEventListeners.add(listener);
+        }
+    }
+
+    @Override
+    public void removeDeviceEventListener(NetconfDeviceOutputEventListener listener) {
+        netconfDeviceEventListeners.remove(listener);
+    }
+
+    @Override
+    public void setEnableNotifications(boolean enableNotifications) {
+        this.enableNotifications = enableNotifications;
+    }
+}