Move internal classes under .impl package (1/2)
Change-Id: I72ed933ecd5ae7ffb268967f80fb395e90f7d8e9
diff --git a/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/impl/NetconfStreamThread.java b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/impl/NetconfStreamThread.java
new file mode 100644
index 0000000..212a4e6
--- /dev/null
+++ b/protocols/netconf/ctl/src/main/java/org/onosproject/netconf/ctl/impl/NetconfStreamThread.java
@@ -0,0 +1,280 @@
+/*
+ * 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.impl;
+
+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.
+ */
+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;
+ }
+}