Thread web service yang sebenarnya

MANIFEST.in

-include *.txt *.ini *.rst
-recursive-include web_starter *.ico *.png *.css *.gif *.jpg *.pt *.txt *.mak
+include CHANGES.txt production.ini development.ini README.rst
+recursive-include iso8583_web *.py *.png *.css *.pt *.tpl *.mako

README.rst

 Daemon ISO8583
 ==============
-
 Daemon ISO8583 ini bisa sebagai client maupun server. Berikut ini cara
 pemasangannya. Nama paketnya mengandung web karena ia bisa dihubungi melalui
 web service JsonRPC untuk melakukan inquiry, payment, dan reversal.
@@ -21,11 +20,8 @@ Salinlah file konfigurasi::
 
     $ cp iso8583.ini test-pemda.ini
 
-Sesuaikan isi section ``[host_bjb]``. Anda bisa menambahkan host lainnya baik
-bank, pemda, bahkan payment gateway dengan menambahkan awalan ``host_`` pada
-section.
-
-Kemudian lakukan uji coba::
+Sesuaikan isi section ``[host_bjb]``. Anda bisa menambahkan host lainnya
+dengan menambahkan awalan ``host_`` pada section. Kemudian lakukan uji coba::
 
     $ ../env/bin/pip install -e '.[testing]'
     $ ../env/bin/pytest iso8583_web/test-conf.py
@@ -48,9 +44,6 @@ Biarkan ini aktif. Buka konsol lain. Sekarang kita buat simulator BJB::
 
 Ubah file ``test-bjb.ini`` pada baris::
 
-    [web]
-    port = 6543
-
     [host_bjb]
     ip = 127.0.0.1
     port = 10002
@@ -59,9 +52,6 @@ Ubah file ``test-bjb.ini`` pada baris::
 
 menjadi::
 
-    [web]
-    port = 6544
-
     [host_pemda]
     ip = 127.0.0.1
     port = 10002
@@ -71,3 +61,79 @@ menjadi::
 Simpan, lalu jalankan::
 
     ../env/bin/iso8583 test-bjb.ini
+
+Log File
+--------
+Untuk menyimpan log ke dalam file lakukan perubahan pada file konfigurasi (INI
+file). Pada section ``[handlers]`` baris ``keys`` tambahkan ``file`` sehingga
+menjadi::
+
+  [handlers]
+  keys = console, file
+
+Pada section ``[logger_root]`` baris ``handlers`` tambahkan ``file`` sehingga
+menjadi::
+
+  [logger_root]
+  level = INFO
+  handlers = console, file
+
+Setelah section ``[handler_console]`` tambahkan section ``[handler_file]``
+seperti ini::
+
+  [handler_file]
+  class = FileHandler
+  args = ('/home/sugiana/tmp/pemda.log', 'a')
+  level = NOTSET
+  formatter = generic
+
+Lalu jalankan lagi daemon seperti di atas.
+
+JsonRpc
+-------
+Ini adalah web service untuk memudahkan pembuatan client. Misalkan kita ingin
+membuat aplikasi teller bank. Pada ``test-bjb.ini`` aktifkan section
+``[web]``::
+
+  [web]
+  port = 7000 
+
+Kemudian restart daemon-nya. Setelah *echo established* dengan daemon pemda
+lakukan *echo request* dengan cara::
+
+  $ ../env/bin/python contrib/bank-teller.py
+
+Hasilnya menjadi seperti ini::
+
+  {'id': 0,
+   'jsonrpc': '2.0',
+   'result': {'code': 0,
+              'data': {'11': '163105',
+                       '39': '00',
+                       '7': '0221163106',
+                       '70': '301'},
+              'message': 'OK'}}
+
+JsonRpc Log File
+----------------
+Biasanya tahap awal pembuatan web client terjadi kesalahan yang tampak di
+client namun tak nampak di log web server. Untuk mengatasi hal ini lakukan
+penambahan di file konfigurasi section ``[loggers]`` menjadi seperti ini::
+
+  [loggers]
+  keys = root, iso8583_web, jsonrpc
+
+Lalu tambahkan section ``[logger_jsonrpc]`` berikut ini::
+
+  [logger_jsonrpc]
+  level = DEBUG
+  handlers = console, file
+  qualname = pyramid_rpc.jsonrpc
+
+Kemudian jalankan lagi daemon-nya. Misalkan client memanggil method ``eco``
+yang seharusnya ``echo`` maka di log akan tampil seperti ini::
+
+  2019-02-21 14:00:47,135 DEBUG handling id:0 method:eco
+  2019-02-21 14:00:47,135 DEBUG json-rpc method not found rpc_id:0 "eco"
+
+Semoga dipahami.

alembic/env.py

@@ -64,6 +64,7 @@ def run_migrations_online():
         with context.begin_transaction():
             context.run_migrations()
 
+
 if context.is_offline_mode():
     run_migrations_offline()
 else:

alembic/versions/074b33635316_alter_users_date_fields_with_time_zone.py

 """alter users date fields with time zone
 
 Revision ID: 074b33635316
-Revises: 
+Revises:
 Create Date: 2018-10-11 12:00:48.568483
 
 """
@@ -17,13 +17,16 @@ depends_on = None
 
 
 def upgrade():
-    op.alter_column('users', 'last_login_date',
+    op.alter_column(
+        'users', 'last_login_date',
         type_=sa.DateTime(timezone=True),
         existing_type=sa.DateTime(timezone=False))
-    op.alter_column('users', 'registered_date',
+    op.alter_column(
+        'users', 'registered_date',
         type_=sa.DateTime(timezone=True),
         existing_type=sa.DateTime(timezone=False))
-    op.alter_column('users', 'security_code_date',
+    op.alter_column(
+        'users', 'security_code_date',
         type_=sa.DateTime(timezone=True),
         existing_type=sa.DateTime(timezone=False))
 

contrib/bank-teller.py

+import requests
+import json
+from pprint import pprint
+
+
+url = 'http://localhost:7000/rpc'
+headers = {'content-type': 'application/json'}
+p = {'host': 'pemda'}
+data = {
+    'method': 'echo',
+    'params': [p],
+    'jsonrpc': '2.0',
+    'id': 0,
+    }
+resp = requests.post(url, data=json.dumps(data), headers=headers)
+json_resp = resp.json()
+pprint(json_resp)

iso8583.ini

@@ -30,8 +30,10 @@ formatter = generic
 [formatter_generic]
 format = %(asctime)s %(levelname)s %(message)s
 
-[web]
-port = 6543
+# Aktifkan JsonRPC server jika ingin simulasi sebagai bank dimana inquiry dkk
+# bisa dilakukan melalui JsonRPC client. 
+# [web]
+# port = 7000 
 
 [host_bjb]
 ip = 127.0.0.1

iso8583_web/read_conf.py

@@ -13,6 +13,7 @@ except ImportError:
 
 listen_ports = []
 ip_conf = {}
+name_conf = {}
 allowed_ips = []
 web = {} 
 
@@ -23,7 +24,7 @@ def get_conf(ip, port):
 
 
 def get_web_port():
-    return web['port']
+    return 'port' in web and web['port']
 
 
 def get_str(conf, section, option, default):
@@ -94,7 +95,7 @@ def read_conf(conf_file):
         cfg = dict()
         cfg['ip'] = ip
         cfg['port'] = port
-        cfg['name'] = section.split('_')[1]
+        cfg['name'] = name = section.split('_')[1]
         cfg['streamer'] = get_str(conf, section, 'streamer', 'none')
         cfg['streamer_cls'] = get_streamer_class(cfg['streamer'])
         cfg['module'] = get_str(conf, section, 'module', 'opensipkd.iso8583.network')
@@ -109,4 +110,4 @@ def read_conf(conf_file):
                 listen_ports.append(port)
             if ip not in allowed_ips:
                 allowed_ips.append(ip)
-        ip_conf[ip_port] = dict(cfg)
+        ip_conf[ip_port] = name_conf[name] = dict(cfg)

iso8583_web/scripts/forwarder.py

@@ -2,12 +2,15 @@ import os
 import sys
 import logging
 import signal
-from time import sleep
+from time import (
+    sleep,
+    time,
+    )
 from threading import Thread
 from wsgiref.simple_server import make_server
 from pyramid.config import Configurator
-from pyramid.response import Response
 from pyramid.paster import setup_logging
+from pyramid_rpc.jsonrpc import jsonrpc_method
 from opensipkd.tcp.connection import (
     ConnectionManager as BaseConnectionManager,
     join_ip_port,
@@ -21,6 +24,11 @@ from opensipkd.tcp.server import (
     RequestHandler as BaseRequestHandler,
     )
 from opensipkd.tcp.client import Client as BaseClient
+from opensipkd.jsonrpc.exc import (
+    JsonRpcInvalidParams,
+    JsonRpcBankNotFound,
+    JsonRpcBillerNetwork,
+    )
 from ..read_conf import (
     read_conf,
     ip_conf,
@@ -28,6 +36,7 @@ from ..read_conf import (
     listen_ports,
     allowed_ips,
     get_conf,
+    name_conf,
     )
 
 
@@ -295,6 +304,15 @@ class Parser(Log):
             self.log_encode(iso)
             raw = iso.getRawIso()
             self.connection.send(raw)
+        else:  # dapat response
+            ip_port = join_ip_port(self.conf['ip'], self.conf['port'])
+            if ip_port in web_process:
+                stan_list = web_process[ip_port]
+                stan = from_iso.get_stan() 
+                if stan in stan_list:
+                    i = stan_list.index(stan)
+                    del stan_list[i]
+                    append_web_response(ip_port, stan, from_iso)
         self.running = False
 
 
@@ -308,18 +326,139 @@ class ConnectionManager(BaseConnectionManager):
 
 
 conn_mgr = ConnectionManager()
-    
+ 
+
 #######
 # Web #
 #######
+def log_web_msg(s):
+    return 'Web server {}'.format(s)
+
+
 def log_web_info(s):
+    msg = log_web_msg(s)
     log = get_log()
-    msg = 'Web server {}'.format(s)
     log.info(msg)
 
 
-def hello_world(request):
-    return Response('Hello World!')
+def log_web_error(s):
+    msg = log_web_msg(s)
+    log = get_log()
+    log.error(msg)
+
+
+def conn_by_name(name):
+    conf = name_conf[name]
+    found_conn = None
+    for ip_port, conn in conn_mgr:
+        ip, port = ip_port.split(':')
+        if conf['ip'] != ip:
+            continue
+        port = int(port)
+        if conf['port'] != port:
+            continue
+        found_conn = conn
+    if not found_conn:
+        raise JsonRpcBankNotFound()
+    if not found_conn.running:
+        raise JsonRpcBankNotFound(message='Disconnected')
+    return found_conn
+
+
+# Daftar job dari web request, berisi iso request
+# key: ip:port, value: list of iso
+web_request = {}
+
+# Daftar job yang sedang diproses, yaitu menunggu iso response
+# key: ip:port, value: list of stan (bit 11) 
+web_process = {}
+
+# Daftar job yang sudah selesai, berisi iso response
+# key: ip:port, value: dict of (key: stan, value: iso)
+web_response = {}
+
+
+def append_web_process(ip_port, iso):
+    stan = iso.get_stan()
+    if ip_port in web_process:
+        web_process[ip_port].append(stan)
+    else:
+        web_process[ip_port] = [stan]
+
+
+def append_web_response(ip_port, stan, iso):
+    if ip_port in web_response:
+        web_response[ip_port][stan] = iso
+    else:
+        web_response[ip_port] = {stan: iso}
+
+
+def web_job(conn, iso):
+    ip_port = join_ip_port(conn.conf['ip'], conn.conf['port'])
+    if ip_port in web_request:
+        web_request[ip_port].append(iso)
+    else:
+        web_request[ip_port] = [iso]
+    stan = iso.get_stan()
+    awal = time()
+    while True:
+        sleep(1)
+        if time() - awal > 5:
+            raise JsonRpcBillerNetwork(message='Timeout')
+        if ip_port not in web_response:
+            continue
+        result = web_response[ip_port]
+        if stan in result:
+            iso = result[stan]
+            del result[stan] 
+            data = iso_to_dict(iso)
+            return dict(code=0, message='OK', data=data)
+
+
+def validate_rpc(p):
+    if 'host' not in p:
+        raise JsonRpcInvalidParams()
+    return conn_by_name(p['host'])
+
+
+def log_web_receive(request, method, p, flow='Receive'):
+    msg = '{} {} {} {}'.format(request.client_addr, flow, method, p)
+    log_web_info(msg)
+
+
+def log_web_send(request, method, p):
+    log_web_receive(request, method, p, 'Send')
+
+
+@jsonrpc_method(endpoint='rpc')
+def echo(request, p):
+    log_web_receive(request, 'echo', p)
+    conn = validate_rpc(p)
+    iso = conn.job.echo_request()
+    r = web_job(conn, iso)
+    log_web_send(request, 'echo', r)
+    return r
+
+
+@jsonrpc_method(endpoint='rpc')
+def inquiry(request, p):
+    conn = validate_rpc(p)
+    iso = conn.job.inquiry(p)
+    return web_job(conn, iso)
+
+
+@jsonrpc_method(endpoint='rpc')
+def payment(request, p):
+    conn = validate_rpc(p)
+    iso = conn.job.payment(p)
+    return web_job(conn, iso)
+
+
+@jsonrpc_method(endpoint='rpc')
+def reversal(request, p):
+    conn = validate_rpc(p)
+    iso = conn.job.reversal(p)
+    return web_job(conn, iso)
 
 
 web_server = {}
@@ -327,18 +466,24 @@ web_server = {}
 
 def start_web_server():
     port = get_web_port()
+    if not port:
+        return
     with Configurator() as config:
-        config.add_route('hello', '/')
-        config.add_view(hello_world, route_name='hello')
+        config.include('pyramid_tm')
+        config.include('pyramid_rpc.jsonrpc')
+        config.add_jsonrpc_endpoint('rpc', '/rpc')
+        config.scan(__name__)
         app = config.make_wsgi_app()
         web_server['listener'] = server = make_server('0.0.0.0', port, app)
-    web_server['thread'] = Thread(target=server.serve_forever) 
+    web_server['thread'] = create_thread(server.serve_forever)
     web_server['thread'].start()
     ip_port = web_server['listener'].server_address
     log_web_info('listen at {}:{}'.format(*ip_port))
 
 
 def stop_web_server(reason):
+    if 'listener' not in web_server:
+        return
     msg = 'stop because {}'.format(reason)
     log_web_info(msg)
     # shutdown() ini kadang tidak segera mengakhiri web server. Akan cepat
@@ -401,7 +546,14 @@ def check_job():
             continue
         iso = connection.job.get_iso()
         if not iso:
-            continue
+            if ip_port not in web_request:
+                continue
+            jobs = web_request[ip_port]
+            if not jobs:
+                continue
+            iso = jobs[0]
+            del jobs[0]
+            append_web_process(ip_port, iso)
         connection.log_encode(iso)
         raw = iso.getRawIso()
         connection.send(raw)

setup.py

@@ -30,12 +30,14 @@ requires = [
     'pyramid_beaker',
     'pyramid_mailer',
     'requests',
+    'pyramid_rpc',
 ]
 
 customs_require = [
     'http://repo.opensipkd.com/pip/opensipkd-base-0.2.tar.gz',
     'http://repo.opensipkd.com/pip/opensipkd-hitung-0.1.tar.gz',
     'http://repo.opensipkd.com/pip/opensipkd-iso8583-0.1.tar.gz',
+    'http://repo.opensipkd.com/pip/opensipkd-jsonrpc-0.1.tar.gz',
     ]
 
 tests_require = [