examples/ping_pong_ns/ping_pong_ns/ping.py

   1 #
   2 #   Copyright 2016 RIFT.IO Inc
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16
  17 from datetime import date
  18 import logging
  19 import json
  20 import socket
  21 import threading
  22 import time
  23
  24 import tornado.web
  25
  26 from util.util import get_url_target
  27
  28 class Ping(object):
  29     def __init__(self):
  30         self._log = logging.getLogger("ping")
  31         self._log.setLevel(logging.DEBUG)
  32
  33         self._ping_count = 0;
  34         self._request_count = 0;
  35         self._response_count = 0;
  36
  37         self._pong_ip = None
  38         self._pong_port = None
  39
  40         self._send_rate = 1 # per second
  41
  42         self._close_lock = threading.Lock()
  43
  44         self._enabled = False
  45         self._socket = None
  46
  47     @property
  48     def rate(self):
  49         return self._send_rate
  50
  51     @rate.setter
  52     def rate(self, value):
  53         self._log.debug("new rate: %s" % value)
  54         self._send_rate = value
  55
  56     @property
  57     def pong_port(self):
  58         return self._pong_port
  59
  60     @pong_port.setter
  61     def pong_port(self, value):
  62         self._log.debug("new pong port: %s" % value)
  63         self._pong_port = value
  64
  65     @property
  66     def pong_ip(self):
  67         return self._pong_ip
  68
  69     @pong_ip.setter
  70     def pong_ip(self, value):
  71
  72         self._log.debug("new pong ip: %s" % value)
  73         self._pong_ip = value
  74
  75     @property
  76     def enabled(self):
  77         return self._enabled
  78
  79     @property
  80     def request_count(self):
  81         return self._request_count
  82
  83     @property
  84     def response_count(self):
  85         return self._response_count
  86
  87     def start(self):
  88         self._log.debug("starting")
  89         self._enabled = True
  90         # self.open_socket()
  91         self.send_thread = threading.Thread(target=self.send_ping)
  92         self.recv_thread = threading.Thread(target=self.recv_resp)
  93         self.send_thread.start()
  94         self.recv_thread.start()
  95
  96     def stop(self):
  97         self._log.debug("stopping")
  98         self._enabled = False
  99         self.close_socket("stopping")
 100
 101     def close_socket(self, msg):
 102         self._close_lock.acquire()
 103         if self._socket != None:
 104             self._socket.close()
 105             self._socket = None
 106             self._log.info("Closed socket with msg={}".format(msg))
 107         self._close_lock.release()
 108
 109     def open_socket(self):
 110         try:
 111             self._log.debug("construct socket")
 112             self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 113             self._socket.settimeout(1)
 114         except socket.error as msg:
 115             self._log.error("error constructing socket %s" % msg)
 116             self._socket = None
 117
 118         while self._enabled:
 119             try:
 120                 self._log.info("Trying to connect....")
 121                 self._socket.connect((self.pong_ip, self.pong_port))
 122                 self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 123                 self._log.info("Socket connected")
 124                 break
 125             except socket.error as msg:
 126                 time.sleep(1)
 127
 128
 129     def send_ping(self):
 130         self.open_socket()
 131
 132         while self._enabled:
 133             if self._socket != None:
 134                 req = "rwpingpong-{}".format(self._ping_count)
 135                 try:
 136                     self._log.info("sending: %s" %req)
 137                     self._socket.sendall(req)
 138                     self._ping_count += 1
 139                     self._request_count += 1
 140                 except socket.error as msg:
 141                     self._log.error("Error({}) sending data".format(msg))
 142                     self.close_socket(msg)
 143                     return
 144
 145             time.sleep(1.0/self._send_rate)
 146
 147         self._log.info("Stopping send_ping")
 148
 149     def recv_resp(self):
 150         while self._enabled:
 151             respb = None
 152             if self._socket != None:
 153                 try:
 154                     respb = self._socket.recv(1024)
 155                 except socket.timeout:
 156                     continue
 157                 except socket.error as msg:
 158                     self._log.error("Error({}) receiving data".format(msg))
 159                     time.sleep(1)
 160                     continue
 161                     # self.close_socket(msg)
 162                     # return
 163
 164             if not respb:
 165                 continue
 166
 167             resp = respb.decode('UTF-8')
 168             self._response_count += 1
 169             self._log.info("receive: %s" % resp)
 170
 171         self._log.info("Stopping recv_resp")
 172
 173 class PingServerHandler(tornado.web.RequestHandler):
 174     def initialize(self, ping_instance):
 175         self._ping_instance = ping_instance
 176
 177     def get(self, args):
 178         response = {'ip': self._ping_instance.pong_ip,
 179                     'port': self._ping_instance.pong_port}
 180
 181         self.write(response)
 182
 183     def post(self, args):
 184         target = get_url_target(self.request.uri)
 185         body = self.request.body.decode("utf-8")
 186         body_header = self.request.headers.get("Content-Type")
 187
 188         if "json" not in body_header:
 189             self.write("Content-Type must be some kind of json 2")
 190             self.set_status(405)
 191             return
 192
 193         try:
 194             json_dicts = json.loads(body)
 195         except:
 196             self.write("Content-Type must be some kind of json 1")
 197             self.set_status(405)
 198             return
 199
 200         if target == "server":
 201             if type(json_dicts['port']) is not int:
 202                 self.set_status(405)
 203                 return
 204
 205             if type(json_dicts['ip']) not in (str, unicode):
 206                 self.set_status(405)
 207                 return
 208
 209             self._ping_instance.pong_ip = json_dicts['ip']
 210             self._ping_instance.pong_port = json_dicts['port']
 211
 212         else:
 213             self.set_status(404)
 214             return
 215
 216         self.set_status(200)
 217
 218 class PingAdminStatusHandler(tornado.web.RequestHandler):
 219     def initialize(self, ping_instance):
 220         self._ping_instance = ping_instance
 221
 222     def get(self, args):
 223         target = get_url_target(self.request.uri)
 224         if target == "state":
 225             value = "enabled" if self._ping_instance.enabled else "disabled"
 226
 227             response = { 'adminstatus': value }
 228         else:
 229             self.set_status(404)
 230             return
 231
 232         self.write(response)
 233
 234     def post(self, args):
 235         target = get_url_target(self.request.uri)
 236         body = self.request.body.decode("utf-8")
 237         body_header = self.request.headers.get("Content-Type")
 238
 239         if "json" not in body_header:
 240             self.write("Content-Type must be some kind of json 2")
 241             self.set_status(405)
 242             return
 243
 244         try:
 245             json_dicts = json.loads(body)
 246         except:
 247             self.write("Content-Type must be some kind of json 1")
 248             self.set_status(405)
 249             return
 250
 251         if target == "state":
 252             if type(json_dicts['enable']) is not bool:
 253                 self.set_status(405)
 254                 return
 255
 256             if json_dicts['enable']:
 257                 if not self._ping_instance.enabled:
 258                     self._ping_instance.start()
 259             else:
 260                 self._ping_instance.stop()
 261
 262         else:
 263             self.set_status(404)
 264             return
 265
 266         self.set_status(200)
 267
 268 class PingStatsHandler(tornado.web.RequestHandler):
 269     def initialize(self, ping_instance):
 270         self._ping_instance = ping_instance
 271
 272     def get(self):
 273         response = {'ping-request-tx-count': self._ping_instance.request_count,
 274                     'ping-response-rx-count': self._ping_instance.response_count}
 275
 276         self.write(response)
 277
 278 class PingRateHandler(tornado.web.RequestHandler):
 279     def initialize(self, ping_instance):
 280         self._ping_instance = ping_instance
 281
 282     def get(self, args):
 283         response = { 'rate': self._ping_instance.rate }
 284
 285         self.write(response)
 286
 287     def post(self, args):
 288         target = get_url_target(self.request.uri)
 289         body = self.request.body.decode("utf-8")
 290         body_header = self.request.headers.get("Content-Type")
 291
 292         if "json" not in body_header:
 293             self.set_status(405)
 294             return
 295
 296         try:
 297             json_dicts = json.loads(body)
 298         except:
 299             self.set_status(405)
 300             return
 301
 302         if target == "rate":
 303             if type(json_dicts['rate']) is not int:
 304                 self.set_status(405)
 305                 return
 306
 307             self._ping_instance.rate = json_dicts['rate']
 308         else:
 309             self.set_status(404)
 310             return
 311
 312         self.set_status(200)