sockets

Author: fikipollo

.gitignore

@@ -1,4 +1,5 @@
 .idea/
 *.pyc
+conf/
 conf/*.cfg
 venv

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2016 Rafa Hernandez <https://github.com/fikipollo>
+Copyright (c) 2019 Rafa Hernandez <https://github.com/fikipollo>
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

VERSION

@@ -1,7 +1,10 @@
-v0.2 December 2018 PySiCa, a simple Python Cache system
+v0.3 January 2019 PySiCa, a simple Python Cache system
 
 CHANGELOG
 
+ - v0.3 January 2019
+    · sd
+
  - v0.2 December 2018
     · Migrated to Python 3
     · Moved to from Flask to Falcon framework (improves performance)

pysica_api.py api/pysica_api_http.pypysica_api.py renamed to api/pysica_api_http.py

@@ -0,0 +1,204 @@
+import socket
+import json
+from struct import unpack, pack
+
+
+class SimpleCache:
+
+    def __init__(self, socket_file=None, server="localhost", port=4444, buffer_size=4096):
+        self.socket_file = socket_file
+        self.server = server
+        self.port = port
+        self.buffer_size = buffer_size
+
+    def add(self, element_id, data, data_type="object", timeout=None, compress=None, user_id=None):
+        return cache_add(element_id, data, data_type, timeout, compress, user_id, socket_file=self.socket_file, server=self.server, port=self.port, buffer_size=self.buffer_size)
+
+    def get(self, element_id=None, data_type=None, user_id=None, reset_timeout=False, timeout=None):
+        return cache_get(element_id, data_type, user_id, reset_timeout, timeout, socket_file=self.socket_file, server=self.server, port=self.port, buffer_size=self.buffer_size)
+
+    def remove(self, element_id, user_id=None, return_elem=False):
+        return cache_remove(element_id, user_id=user_id, return_elem=return_elem, socket_file=self.socket_file, server=self.server, port=self.port, buffer_size=self.buffer_size)
+
+    def reset(self, element_id=None, timeout=None, user_id=None):
+        return cache_reset(element_id, timeout=timeout, user_id=user_id, socket_file=self.socket_file, server=self.server, port=self.port, buffer_size=self.buffer_size)
+
+
+def cache_add(element_id, data, data_type, timeout=None, compress=None, user_id=None, socket_file=None, server="localhost", port=4444, buffer_size=4096):
+    try:
+        cp = SocketHandler()
+        cp.connect(socket_file, server, port)
+        # Prepare package
+        data = {
+            'target': "add",
+            'element_id': element_id,
+            'data': data,
+            'data_type': data_type,
+            'timeout': timeout,
+            'compress': compress,
+            'user_id': user_id
+        }
+        # Send data
+        response = cp.send_data(data, buffer_size=buffer_size)
+        cp.close()
+        return Response(json.loads(response))
+    except:
+        return Response({"success": False, "message": "Unable to connect to cache server."})
+
+
+def cache_get(element_id=None, data_type=None, user_id=None, reset_timeout=False, timeout=None, socket_file=None, server="localhost", port=4444, buffer_size=4096):
+    try:
+        cp = SocketHandler()
+        cp.connect(socket_file, server, port)
+        # Prepare package
+        data = {
+            'target': "get",
+            'element_id': element_id,
+            'data_type': data_type,
+            'user_id': user_id,
+            'reset_timeout': reset_timeout,
+            'timeout': timeout
+        }
+        # Get data
+        response = cp.get_data(data, buffer_size=buffer_size)
+        cp.close()
+        return Response(json.loads(response))
+    except:
+        return Response({"success": False, "message": "Unable to connect to cache server."})
+
+
+def cache_remove(element_id, user_id=None, return_elem=False, socket_file=None, server="localhost", port=4444, buffer_size=4096):
+    try:
+        # Get data
+        if return_elem:
+            element = cache_get(element_id=element_id, user_id=user_id, socket_file=socket_file, server=server, port=port, buffer_size=buffer_size)
+        # Remove element
+        cp = SocketHandler()
+        cp.connect(socket_file, server, port)
+        print("sadasds")
+        # Prepare package
+        data={
+            'target': "remove",
+            'element_id': element_id,
+            'user_id': user_id
+        }
+        response = cp.send_data(data, buffer_size=buffer_size)
+        response = json.loads(response)
+
+        if return_elem:
+            response["result"] = element.result
+
+        cp.close()
+        return Response(response)
+    except:
+        return Response({"success": False, "message": "Unable to connect to cache server."})
+
+
+def cache_reset(element_id, timeout=None, user_id=None, socket_file=None, server="localhost", port=4444, buffer_size=4096):
+    try:
+        cp = SocketHandler()
+        cp.connect(socket_file, server, port)
+        # Prepare package
+        data={
+            'target': "reset",
+            'element_id': element_id,
+            'timeout': timeout,
+            'user_id': user_id
+        }
+        # Get data
+        response = cp.send_data(data, buffer_size=buffer_size)
+        cp.close()
+        return Response(json.loads(response))
+    except:
+        return Response({"success": False, "message": "Unable to connect to cache server."})
+
+
+class SocketHandler:
+    def __init__(self):
+        self.socket = None
+
+    def connect(self, socket_file=None, server_ip=None, server_port=None):
+        if socket_file is not None:
+            self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            self.socket.connect(socket_file)
+        else:
+            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.socket.connect((server_ip, server_port))
+
+    def close(self):
+        self.socket.shutdown(socket.SHUT_WR)
+        self.socket.close()
+        self.socket = None
+
+    def send_data(self, data, buffer_size=4096):
+        data = json.dumps(data)
+        # use struct to make sure we have a consistent endianness on the length
+        length = pack('>Q', len(data))
+        # sendall to make sure it blocks if there's back-pressure on the socket
+        self.socket.sendall(length)
+        self.socket.sendall(data)
+        # Receive respond
+        bs = self.socket.recv(8)
+        (data_length,) = unpack('>Q', bs)
+        data = b''
+        while len(data) < data_length:
+            # doing it in batches is generally better than trying
+            # to do it all in one go, so I believe.
+            to_read = data_length - len(data)
+            data += self.socket.recv(buffer_size if to_read > buffer_size else to_read)
+        return data
+
+    def get_data(self, data, buffer_size=4096):
+        data = json.dumps(data)
+        # use struct to make sure we have a consistent endianness on the length
+        length = pack('>Q', len(data))
+        # sendall to make sure it blocks if there's back-pressure on the socket
+        self.socket.sendall(length)
+        self.socket.sendall(data)
+        # Receive respond
+        bs = self.socket.recv(8)
+        (data_length,) = unpack('>Q', bs)
+        data = b''
+        while len(data) < data_length:
+            # doing it in batches is generally better than trying
+            # to do it all in one go, so I believe.
+            to_read = data_length - len(data)
+            data += self.socket.recv(buffer_size if to_read > buffer_size else to_read)
+        return data
+
+
+class Response(object):
+    def __init__(self, json_object):
+        if "success" in json_object:
+            self.success = json_object.get("success")
+        if "element_id" in json_object:
+            self.element_id = json_object.get("element_id")
+        if "result" in json_object:
+            self.result = json_object.get("result")
+        if "message" in json_object:
+            self.message = json_object.get("message")
+
+    def __str__(self):
+        _str = dict()
+        try:
+            _str["success"] = self.success
+        except:
+            pass
+        try:
+            _str["element_id"] = self.element_id
+        except:
+            pass
+        try:
+            _str["result"] = self.result
+        except:
+            pass
+        try:
+            _str["message"] = self.message
+        except:
+            pass
+        return str(_str)
+
+    def __repr__(self):
+        return self.__str__()
+
+

api/pysica_api_sockets.py

@@ -1,5 +1,8 @@
 {
     "SERVER_SETTINGS" : {
+        "SERVER_MODE" : "web_server",
+        "SERVER_SOCKET_FILE" : "",
+        "SERVER_BUFFER_SIZE" : 4096,
         "SERVER_HOST_NAME" : "0.0.0.0",
         "SERVER_SUBDOMAIN" : "",
         "SERVER_PORT_NUMBER" : 4444,

default/server.default.cfg

@@ -1,14 +1,52 @@
 #!/bin/ash
 
-if [[ -f /tmp/pysica.pid ]]; then
-  kill -9 `cat /tmp/pysica.pid`
-  rm /tmp/pysica.*
-fi
-
+# First create the required directories
 mkdir -p /var/log/pysica
 mkdir -p /var/www/pysica/conf
 touch /var/log/pysica/cache.log
 
-uwsgi --ini /var/pysica_uwsgi.ini
+if [[ ! -f /var/www/pysica/conf/server.cfg ]]; then
+    cp /var/www/pysica/default/logging.default.cfg /var/www/pysica/conf/logging.cfg
+    cp /var/www/pysica/default/server.default.cfg /var/www/pysica/conf/server.cfg
+fi
+
+# Read environment variables
+SERVER_MODE="${SERVER_MODE:-web_server}"
+SERVER_SOCKET_FILE="${SERVER_SOCKET_FILE:-}"
+SERVER_BUFFER_SIZE=${SERVER_BUFFER_SIZE:-4096}
+SERVER_HOST_NAME="${SERVER_HOST_NAME:-0.0.0.0}"
+
+# Change the default configuration based on environment
+sed -i 's/"SERVER_MODE".*/"SERVER_MODE" : "'${SERVER_MODE}'",/' /var/www/pysica/conf/server.cfg
+sed -i 's/"SERVER_SOCKET_FILE".*/"SERVER_SOCKET_FILE" : "'${SERVER_SOCKET_FILE}'",/' /var/www/pysica/conf/server.cfg
+sed -i 's/"SERVER_BUFFER_SIZE".*/"SERVER_BUFFER_SIZE" : '${SERVER_BUFFER_SIZE}',/' /var/www/pysica/conf/server.cfg
+sed -i 's/"SERVER_HOST_NAME".*/"SERVER_HOST_NAME" : "'${SERVER_HOST_NAME}'",/' /var/www/pysica/conf/server.cfg
+
+
+echo "#---------------------------------------------------------------------------------------------------"
+echo "# Welcome to PySiCa (Python Simple Cache)"
+
+# Launch server based on the current configuration
+if [[ "$SERVER_MODE" == "socket_based" ]]; then
+    echo "# Cache server is running in SOCKET mode"
+    echo "# To change the running mode, please set the value for the variable SERVER_MODE in the"
+    echo "# the server.cfg file. Available options are 'web_based' for RESTFUL mode based on HTTP "
+    echo "# and 'socket_based' to use pure socket communication."
+    echo "#---------------------------------------------------------------------------------------------------"
+    python /var/www/pysica/server_sockets.py
+else
+    echo "# Cache server is running as web server"
+    echo "# To change the running mode, please set the value for the variable SERVER_MODE in the"
+    echo "# the server.cfg file. Available options are 'web_based' for RESTFUL mode based on HTTP "
+    echo "# and 'socket_based' to use pure socket communication."
+    echo "#---------------------------------------------------------------------------------------------------"
+
+    if [[ -f /tmp/pysica.pid ]]; then
+      kill -9 `cat /tmp/pysica.pid`
+      rm /tmp/pysica.*
+    fi
+
+    uwsgi --ini /var/pysica_uwsgi.ini
+    tail -f /var/log/pysica/cache.log
+fi
 
-tail -f /var/log/pysica/cache.log

docker/configs/entrypoint.sh

@@ -1,6 +1,6 @@
 [uwsgi]
 chdir           = /var/www/pysica
-wsgi-file       = server.py
+wsgi-file       = server_http.py
 callable        = application
 processes       = 1
 enable-threads  = true

docker/configs/pysica.zip

@@ -4,16 +4,13 @@ services:
     $${INSTANCE_NAME}-cache:
       image: fikipollo/pysica
       container_name: $${INSTANCE_NAME}-cache
-      depends_on:
-        - $${INSTANCE_NAME}-data
-      volumes_from:
-        - $${INSTANCE_NAME}-data
       ports:
-        - "4444:4444"
-
-    $${INSTANCE_NAME}-data:
-      image: busybox
-      container_name: $${INSTANCE_NAME}-data
+        - "$${INSTANCE_PORT_1}:4444"
+      environment:
+        - SERVER_MODE=web_based # web_based or socket_based
+        - SERVER_SOCKET_FILE=/some/file.sock
+        - SERVER_BUFFER_SIZE=4096
+        - SERVER_HOST_NAME=my.cache.com
       volumes:
         - /data/$${INSTANCE_NAME}-data/logs:/var/log/pysica
         - /data/$${INSTANCE_NAME}-data/conf:/var/www/pysica/conf