vnfs/TestVNF/netconfserver/src/main/java/com/ericsson/testvnf/server/netconf/NetconfHandler.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package com.ericsson.testvnf.server.netconf;
  22
  23 import java.io.BufferedReader;
  24 import java.io.InputStream;
  25 import java.io.InputStreamReader;
  26 import java.io.OutputStream;
  27 import java.io.StringReader;
  28 import java.util.Map;
  29 import java.util.stream.Stream;
  30
  31 import org.apache.commons.logging.Log;
  32 import org.apache.commons.logging.LogFactory;
  33 import org.xml.sax.InputSource;
  34 import org.xml.sax.SAXException;
  35 import org.xml.sax.XMLReader;
  36 import org.xml.sax.helpers.XMLReaderFactory;
  37
  38 import com.ericsson.testvnf.server.helper.ActionHelper;
  39 import com.ericsson.testvnf.server.helper.CustomParser;
  40 import com.ericsson.testvnf.server.models.Hello;
  41 import com.ericsson.testvnf.server.models.NetconfMessage;
  42 import com.ericsson.testvnf.server.models.RpcData;
  43 import com.ericsson.testvnf.server.requestqueue.RequestQueueHandler;
  44
  45 /*
  46  * NetconfHandler class which handles the netconf communication with a client
  47  */
  48 public class NetconfHandler implements Runnable {
  49
  50         private InputStream in;
  51         private Map<String, Boolean> connectionResetMap;
  52         private XMLReader xmlReader;
  53         private boolean sessionClosed;
  54         private RequestQueueHandler requestQueueHandler;
  55         private ActionHelper actionHelper;
  56         private Thread requestHandlerThread;
  57         private static final Log log = LogFactory.getLog(NetconfHandler.class);
  58
  59         public NetconfHandler(InputStream in, OutputStream out, Map<String, Boolean> connectionResetMap) {
  60                 this.connectionResetMap = connectionResetMap;
  61                 this.in = in;
  62                 actionHelper = new ActionHelper(out);
  63                 this.sessionClosed = false;
  64         }
  65
  66         public void run() {
  67                 // initialize rpc request handler and request queue
  68                 try {
  69                         requestQueueHandler = new RequestQueueHandler();
  70                         CustomParser customParser = new CustomParser();
  71                         customParser.setRequestQueueHandler(requestQueueHandler); // input requests once processed are added to the request queue.
  72                         xmlReader = XMLReaderFactory.createXMLReader();
  73                         xmlReader.setContentHandler(customParser);
  74                         xmlReader.setErrorHandler(customParser);
  75                 } catch (SAXException e) {
  76                         log.error("Error creating custom rpc request parser.", e);
  77                         return;
  78                 }
  79                 actionHelper.sendHelloMessage(); // server sends hello to the client as soon as the client initiates a connection with the server
  80                 startRequestHandler();
  81                 // start and wait for request handler
  82                 try {
  83                         Thread.sleep(1000);
  84                 } catch (InterruptedException e) {
  85                         Thread.currentThread().interrupt();
  86                         log.warn("Error waiting for thread.", e);
  87                 }
  88                 readInputRequest();
  89
  90         }
  91
  92         // read input requests from the client
  93         private void readInputRequest(){
  94
  95                         StringBuilder netconfMessage = new StringBuilder();
  96                         try (BufferedReader br = new BufferedReader(new InputStreamReader(in));
  97                                         Stream<String> lineStream = br.lines();){
  98                                 log.info("Parsing message.");
  99                                 // rpc requests from clients are parsed and processed
 100                                 lineStream.forEach(ele -> {
 101                                         ele = ele.trim();
 102                                         log.info("current element::" + ele);
 103                                         if (ele.contains("]]>]]>")) {
 104                                                 ele = ele.replace("]]>]]>", "");
 105                                         }
 106                                         if (ele.contains("</hello>") || ele.contains("</rpc>")) {
 107                                                 netconfMessage.append(ele + '\n');
 108                                                 String messageAsString = netconfMessage.toString();
 109                                                 try {
 110                                                         log.info("Parsing message---:\n" + messageAsString);
 111                                                         xmlReader.parse(new InputSource(new StringReader(messageAsString.trim()))); //xmlParser parses the rpc requests
 112                                                         log.info("Parsing done..");
 113                                                 } catch (Exception e) {
 114                                                         log.error("Error parsing. Message---: \n" + messageAsString, e);
 115                                                         sessionClosed = true;
 116                                                 }
 117                                                 netconfMessage.setLength(0);// reset the message as one whole request is complete
 118                                         }else {
 119                                                 netconfMessage.append(ele + '\n');
 120                                         }
 121                                 });
 122                         } catch (Exception e) {
 123                                 log.error("Exception caught in NetconfHandler readInputRequest: "+ e.getMessage());
 124                         } finally {
 125                                 interruptThreads();
 126                         }
 127         }
 128
 129     // method that performs actions based on the message coming in from the client
 130         private void startRequestHandler() {
 131                 log.info("start RequestHandler.");
 132                 requestHandlerThread = new Thread("Request handler") {
 133                         @Override
 134                         public void run() {
 135                                 while (!sessionClosed) {
 136
 137                                         NetconfMessage netconfMessage = null;
 138                                         try {
 139                                                 netconfMessage = requestQueueHandler.waitAndGetMessageFromQueue(); // get the message received
 140                                         } catch (InterruptedException e) {
 141                                                 log.warn("Interrupted exception");
 142                                                 Thread.currentThread().interrupt();
 143                                                 break;
 144                                         }
 145
 146                                         if (netconfMessage instanceof Hello) { // if client sends a hello, send a hello message back
 147                                                 actionHelper.sendHelloMessage();
 148                                         } else if (netconfMessage instanceof RpcData) {
 149                                                 sessionClosed = actionHelper.doActionForRPCRequest(netconfMessage, connectionResetMap, sessionClosed);
 150                                         }else {
 151                                                 log.warn("Unknown message received.");
 152                                         }
 153                                 }
 154                                 log.info("Request handler ended");
 155                         }
 156
 157                 };
 158                 requestHandlerThread.start();
 159                 log.info("Request handler thread started.");
 160         }
 161
 162         public void interruptThreads() {
 163                 actionHelper.interruptGroovyCallerThread();
 164                 if (requestHandlerThread!=null && requestHandlerThread.isAlive()) {
 165                         log.info("Killing request handler thread");
 166                         requestHandlerThread.interrupt();
 167                 }
 168
 169         }
 170
 171 }