Skip to content

Owo Sugiana / sismiop-models

Go to a project
Commit 27eea365 by aa.gustiana@gmail.com
Options

aagusti2 init

1 parent 3de1e0f6
Inline Side-by-side
Showing 8 changed files with 382 additions and 149 deletions
sismiop/services/base.py
......@@ -143,9 +143,13 @@ class Inquiry(Query):
def __init__(
self, invoice_id=None, persen_denda=2, invoice=None,
tgl_bayar=None, debug=False):
import time as _time
t0 = _time.time()
Query.__init__(self, invoice_id, invoice)
t1 = _time.time()
self.debug_invoice = None
if not self.invoice:
self._init_timing = {'Query.__init__': t1-t0}
return
if debug:
self.debug_invoice = self.invoice
......@@ -155,13 +159,28 @@ class Inquiry(Query):
self.tgl_bayar = tgl_bayar
else:
self.tgl_bayar = date.today()
t2 = _time.time()
if not self.is_available():
self.invoice = None
self._init_timing = {'Query.__init__': t1-t0, 'is_available': t2-t1}
return
self.persen_denda = self.get_persen_denda()
# Digunakan untuk ISO8583
self.tagihan = self.denda = self.discount = self.total = 0
t3 = _time.time()
self.hitung()
t4 = _time.time()
self._init_timing = {
'Query.__init__': t1-t0,
'is_available': t2-t1,
'get_persen_denda': t3-t2,
'hitung': t4-t3,
'total': t4-t0
}
try:
from iso8583_web.iso8583 import log
log.info(f"[TIMER] Inquiry.__init__: " + ", ".join(f"{k}={v*1000:.2f}ms" for k,v in self._init_timing.items()))
except Exception:
pass
def get_persen_denda(self):
if self.get_jatuh_tempo().year < 2024:
......
tests/demo.txt 0 → 100644
No preview for this file type
tests/hasil 0 → 100644
File mode changed
tests/hasildetail.txt 0 → 100644
No preview for this file type
tests/max_threads_test.py 0 → 100644
import threading
import time
import os
import sys
def main():
print(f"Logical CPUs: {os.cpu_count()}")
print(f"Python version: {sys.version}")
print("Testing maximum number of threads...")
threads = []
max_threads = 0
try:
while True:
t = threading.Thread(target=time.sleep, args=(10,))
t.start()
threads.append(t)
max_threads += 1
if max_threads % 100 == 0:
print(f"Threads started: {max_threads}")
except Exception as e:
print(f"Max threads reached: {max_threads}")
print(f"Exception: {e}")
finally:
print("Cleaning up threads...")
for t in threads:
t.join(timeout=0.1)
print("Done.")
if __name__ == "__main__":
main()
tests/test-iso8583-v2.py 0 → 100644
from calendar import c
from datetime import datetime, timedelta
import socket
import time
from ISO8583.ISO8583 import ISO8583
import threading
import argparse
from sqlalchemy import false
def get_args():
parser = argparse.ArgumentParser(description='ISO8583 network test')
parser.add_argument('--host', default='localhost', help='Server host (default: localhost)')
parser.add_argument('--port', type=int, default=8585, help='Server port (default: 8583)')
parser.add_argument('--network-count', type=int, default=5, help='Number of network management messages to send (default: 5)')
parser.add_argument('--no-etx', dest='etx', action='store_false', help='Disable ETX trailer (default: enabled)')
parser.add_argument('--etx', dest='etx', action='store_true', help='Enable ETX trailer (default: enabled)')
parser.add_argument('--inv-file', type=str, default=None, help='File with lines for bit61 values (one per transaction)')
parser.set_defaults(etx=True)
return parser.parse_args()
def send_0200_transaction(sock, stan, bit61, use_etx=True):
payload = build_0200_transaction(stan, bit61=bit61)
if isinstance(payload, str):
payload_bytes = payload.encode('ascii')
else:
payload_bytes = payload
if payload_bytes and payload_bytes[0] == 0:
payload_bytes = payload_bytes[1:]
trailer = ETX if use_etx else b''
frame_len = len(payload_bytes) + len(trailer)
length = str(frame_len).zfill(4).encode('ascii')
sock.sendall(length + payload_bytes + trailer)
print(f"[SEND 0200] {payload_bytes}{trailer} (ETX={'on' if use_etx else 'off'})")
# Receive response
# Helper to build a simple 0800 network management request using ISO8583 module
def build_0800_network(stan: str) -> bytes:
iso = ISO8583()
iso.setMTI('0800')
iso.setBit(7, time.strftime('%m%d%H%M%S'))
iso.setBit(11, stan)
iso.setBit(70, '301')
return iso.getRawIso()
class ISO8583NetworkTest():
def __init__(self, sock: socket.socket, use_etx: bool=False):
self.sock = sock
self.use_etx = use_etx
def send_network_test(self, stan):
payload = build_0800_network(stan)
# Ensure payload is ASCII bytes
if isinstance(payload, str):
payload_bytes = payload.encode('ascii')
else:
payload_bytes = payload
# Remove leading null byte if present
if payload_bytes and payload_bytes[0] == 0:
payload_bytes = payload_bytes[1:]
trailer = ETX if self.use_etx else b''
frame_len = len(payload_bytes) + len(trailer)
length = str(frame_len).zfill(4).encode('ascii')
self.sock.sendall(length + payload_bytes + trailer)
ETX = b'\x03'
def build_0200_transaction(stan: str, *, bit61: str) -> bytes:
dt = datetime.now()
iso = ISO8583()
iso.setMTI('0200')
iso.setBit(2, '622011444444444444') # PAN
iso.setBit(3, '341019') # Processing code
iso.setBit(4, '000000000000') # Amount
iso.setBit(7, dt.strftime('%m%d%H%M%S'))
iso.setBit(11, stan)
iso.setBit(12, dt.strftime('%H%M%S'))
iso.setBit(13, dt.strftime('%m%d'))
iso.setBit(15, (dt + timedelta(days=1)).strftime('%m%d'))
iso.setBit(18, '6010')
iso.setBit(22, '021')
iso.setBit(32, '110')
iso.setBit(33, '00110')
iso.setBit(35, '622011444444444444=9912')
iso.setBit(37, ('000000' + stan)[-12:])
iso.setBit(41, 'N703'.ljust(8))
iso.setBit(42, '02N703'.ljust(15))
iso.setBit(43, 'TLR N703'.ljust(40))
iso.setBit(49, '360')
iso.setBit(59, 'PAY')
iso.setBit(60, '120')
iso.setBit(61, bit61)
iso.setBit(63, '214')
iso.setBit(102, '0010823214360'.ljust(20))
iso.setBit(107, '0010')
return iso.getRawIso()
def recv_data(sock, use_etx=True):
while True:
try:
length_raw = sock.recv(4)
except Exception as e:
print(f"[RECV ERROR] Failed to read length: {e}")
continue
if not length_raw:
continue
resp_len = int(length_raw.decode('ascii'))
resp = b''
while len(resp) < resp_len:
chunk = sock.recv(resp_len - len(resp))
if not chunk:
raise ConnectionError('Socket closed while reading response')
resp += chunk
print(f"[RECV 0200] {resp}")
if use_etx and resp.endswith(ETX):
resp = resp[:-1]
iso_resp = ISO8583()
iso_resp.setIsoContent(resp)
print(f"[0200 RESP] MTI={iso_resp.getMTI()} Bit39={iso_resp.getBit(39)}")
def main():
args = get_args()
host = args.host
port = args.port
with socket.create_connection((host, port), timeout=10) as sock:
threads = []
for i in range(args.network_count):
stan = str(100000 + i)[-6:]
server = ISO8583NetworkTest(sock=sock, use_etx=args.etx)
t_send = threading.Thread(target=server.send_network_test, args=(stan,))
t_send.start()
threads.append(t_send)
for t_send in threads:
t_send.join()
def receiver(sock, use_etx):
recv_data(sock, use_etx=use_etx)
t_recv = threading.Thread(target=receiver, args=(sock, args.etx))
t_recv.start()
t_recv.join()
if args.inv_file:
# Each line in inv-file is sent in its own thread, each with its own socket
with open(args.inv_file, 'r') as f:
bit61_lines = [line.strip() for line in f if line.strip()]
threads = []
def send_0200_thread(stan, bit61, use_etx, host, port):
send_0200_transaction(sock, stan, bit61, use_etx)
for i, bit61 in enumerate(bit61_lines):
if i<2:
stan = str(100000 + i)[-6:]
t = threading.Thread(target=send_0200_thread, args=(stan, bit61, args.etx, args.host, args.port))
t.start()
threads.append(t)
for t in threads:
t.join()
if __name__ == '__main__':
main()
tests/test-iso8583-v2py.py 0 → 100644
from ISO8583.ISO8583 import ISO8583
tests/test-iso8583_send.py
......@@ -5,6 +5,7 @@ profile (field 61 / bit61).
Example:
python test-iso8583_send.py --host localhost --port 8592 --inv-id 3275050002008001801990
python test-iso8583_send.py --host 36.95.7.36 --port 8595 --inv-file invoice_bks.txt --threads 50 --timeout 30 --no-login --no-etx
"""
from __future__ import annotations
......@@ -16,7 +17,7 @@ import time
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from ISO8583.ISO8583 import ISO8583
ETX = b"\x03"
......@@ -99,7 +100,7 @@ def recv_frame(sock: socket.socket, *, expect_etx: bool) -> bytes:
except ValueError as e:
raise ValueError(f"invalid length prefix: {length_raw!r}") from e
data = recv_exact(sock, length)
data = recv_exact(sock, length-4)
if expect_etx:
if not data.endswith(ETX):
raise ValueError(f"invalid frame trailer (expected ETX 0x03): {data[-1:]!r}")
......@@ -181,20 +182,27 @@ _print_lock = threading.Lock()
def log_line(line: str) -> None:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
line = f"{now} {line}"
with _print_lock:
print(line, flush=True)
def format_ok_line(*, kind: str, inv_id: Optional[str], rc39: Optional[str], send_ts: str, t0: float) -> str:
def format_ok_line(*, kind: str, inv_id: Optional[str], rc39: Optional[str], send_ts: str, t0: float, stan: Optional[str] = None) -> str:
dt_ms = (time.perf_counter() - t0) * 1000.0
inv_part = "" if inv_id is None else f" inv_id={inv_id}"
return f"{send_ts} {kind}{inv_part} rc39={rc39} duration_ms={dt_ms:.2f}"
stan_part = f" stan={stan}" if stan is not None else ""
dt = datetime.strptime(send_ts, '%Y-%m-%d %H:%M:%S.%f')
akhir= datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
return f"{send_ts} {kind}{inv_part}{stan_part} rc39={rc39} duration_ms={dt_ms:.2f} elapsed={akhir}s"
def format_err_line(*, kind: str, inv_id: Optional[str], exc: BaseException, send_ts: str, t0: float) -> str:
def format_err_line(*, kind: str, inv_id: Optional[str], exc: BaseException, send_ts: str, t0: float, stan: Optional[str] = None) -> str:
dt_ms = (time.perf_counter() - t0) * 1000.0
inv_part = "" if inv_id is None else f" inv_id={inv_id}"
return f"{send_ts} {kind}{inv_part} ERROR {type(exc).__name__}:{exc} duration_ms={dt_ms:.2f}"
stan_part = f" stan={stan}" if stan is not None else ""
return f"{send_ts} {kind}{inv_part}{stan_part} ERROR {type(exc).__name__}:{exc} duration_ms={dt_ms:.2f}"
@dataclass
......@@ -219,165 +227,94 @@ class _Pending:
error: Optional[str] = None
class StanDispatcher:
"""Single TCP connection dispatcher keyed by STAN (field 11).
- Many threads may call .tx() concurrently.
- Only one receiver thread reads from socket.
- Responses are routed to the right waiter using bit 11 (STAN).
"""
def __init__(self, sock: socket.socket, *, use_etx: bool, timeout_s: float):
self._sock = sock
self._use_etx = use_etx
self._timeout_s = timeout_s
# Standalone receipt thread for STAN dispatch
def log_send(kind: str, inv_id: str|None, stan: str, send_ts: str):
inv_part = f" inv_id={inv_id}" if inv_id else ""
log_line(f"SEND {send_ts} {kind}{inv_part} stan={stan}")
self._closed = threading.Event()
self._send_lock = threading.Lock()
self._pending_lock = threading.Lock()
self._pending: Dict[str, _Pending] = {}
def log_receipt(kind: str, inv_id: str|None, stan: str, rc39: str|None, recv_ts: str):
inv_part = f" inv_id={inv_id}" if inv_id else ""
log_line(f"RECEIPT {recv_ts} {kind}{inv_part} stan={stan} rc39={rc39}")
self._stan_lock = threading.Lock()
self._stan_counter = StanCounter()
class ReceiptThread(threading.Thread):
# def log_send(kind: str, inv_id: str|None, stan: str, send_ts: str):
# inv_part = f" inv_id={inv_id}" if inv_id else ""
# log_line(f"SEND {send_ts} {kind}{inv_part} stan={stan}")
# Receiver loop uses a short timeout to notice close.
self._sock.settimeout(1.0)
self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True)
self._rx_thread.start()
def close(self) -> None:
self._closed.set()
try:
self._sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
self._sock.close()
except OSError:
pass
# Unblock any waiters.
with self._pending_lock:
pendings = list(self._pending.values())
self._pending.clear()
for p in pendings:
p.error = p.error or "closed"
if p.kind == "TX":
p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
else:
p.ok = False
p.event.set()
def _next_stan(self) -> str:
with self._stan_lock:
return self._stan_counter.next()
def _fail_all(self, err: BaseException) -> None:
with self._pending_lock:
pendings = list(self._pending.values())
self._pending.clear()
for p in pendings:
p.error = type(err).__name__
if p.kind == "TX":
p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
else:
p.ok = False
p.event.set()
# def log_receipt(kind: str, inv_id: str|None, stan: str, rc39: str|None, recv_ts: str):
# inv_part = f" inv_id={inv_id}" if inv_id else ""
# log_line(f"RECEIPT {recv_ts} {kind}{inv_part} stan={stan} rc39={rc39}")
def __init__(self, sock, use_etx, pending, pending_lock, closed):
super().__init__(daemon=True)
self.sock = sock
self.use_etx = use_etx
self.pending = pending
self.pending_lock = pending_lock
self.closed = closed
self.sock.settimeout(1.0)
def _rx_loop(self) -> None:
while not self._closed.is_set():
def run(self):
while not self.closed.is_set():
try:
resp = recv_frame(self._sock, expect_etx=self._use_etx)
resp = recv_frame(self.sock, expect_etx=self.use_etx)
except socket.timeout:
continue
except (OSError, ValueError) as e:
self._fail_all(e)
with self.pending_lock:
pendings = list(self.pending.values())
self.pending.clear()
for p in pendings:
p.error = type(e).__name__
if p.kind == "TX":
p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
else:
p.ok = False
p.event.set()
return
# Print/log the raw response bytes received
log_line(f"[RECEIPT RAW] {resp.hex()}")
try:
parsed = parse_until_39(resp)
except (ValueError, UnicodeError) as e:
self._fail_all(e)
with self.pending_lock:
pendings = list(self.pending.values())
self.pending.clear()
for p in pendings:
p.error = type(e).__name__
if p.kind == "TX":
p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
else:
p.ok = False
p.event.set()
return
stan = parsed.get(11)
if not stan:
continue
with self._pending_lock:
p = self._pending.pop(stan, None)
with self.pending_lock:
p = self.pending.pop(stan, None)
if p is None:
continue
rc = parsed.get(39)
dt_ms = (time.perf_counter() - p.t0) * 1000.0
recv_ts = ts()
log_receipt(p.kind, p.inv_id, stan, rc, recv_ts)
if p.kind == "SIGNON":
ok = parsed.get(0) == "0810" and rc == "00"
log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=p.send_ts, t0=p.t0))
log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=p.send_ts, t0=p.t0, stan=stan))
p.ok = ok
else:
ok = parsed.get(0) == "0210" and rc == "00"
log_line(format_ok_line(kind="TX", inv_id=p.inv_id, rc39=rc, send_ts=p.send_ts, t0=p.t0))
log_line(format_ok_line(kind="TX", inv_id=p.inv_id, rc39=rc, send_ts=p.send_ts, t0=p.t0, stan=stan))
p.result = TxResult(inv_id=p.inv_id or "", ok=ok, rc39=rc, duration_ms=dt_ms)
p.event.set()
def signon(self) -> bool:
stan = self._next_stan()
payload = build_0800_network(stan, "301")
send_ts = ts()
t0 = time.perf_counter()
p = _Pending(kind="SIGNON", inv_id=None, stan=stan, send_ts=send_ts, t0=t0, event=threading.Event())
with self._pending_lock:
self._pending[stan] = p
try:
with self._send_lock:
send_frame(self._sock, payload, use_etx=self._use_etx)
except (OSError, ValueError) as e:
with self._pending_lock:
self._pending.pop(stan, None)
log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0))
return False
if not p.event.wait(self._timeout_s):
with self._pending_lock:
self._pending.pop(stan, None)
log_line(format_err_line(kind="SIGNON", inv_id=None, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0))
return False
return bool(p.ok)
def tx(self, inv_id: str) -> TxResult:
stan = self._next_stan()
payload = build_0200_transaction(stan, bit61=inv_id)
send_ts = ts()
t0 = time.perf_counter()
p = _Pending(kind="TX", inv_id=inv_id, stan=stan, send_ts=send_ts, t0=t0, event=threading.Event())
with self._pending_lock:
self._pending[stan] = p
try:
with self._send_lock:
send_frame(self._sock, payload, use_etx=self._use_etx)
except (OSError, ValueError) as e:
with self._pending_lock:
self._pending.pop(stan, None)
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0))
return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__)
if not p.event.wait(self._timeout_s):
with self._pending_lock:
self._pending.pop(stan, None)
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0))
return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error="timeout")
return p.result or TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=p.error or "unknown")
def send_signon(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounter") -> bool:
"""Minimal sign-on: send 0800/301 and check for 0810/00 response."""
stan = stan_counter.next()
payload = build_0800_network(stan, "301")
send_ts = ts()
......@@ -387,10 +324,10 @@ def send_signon(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounte
resp = recv_frame(sock, expect_etx=use_etx)
parsed = parse_until_39(resp)
rc = parsed.get(39)
log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=send_ts, t0=t0))
log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=send_ts, t0=t0, stan=stan))
return parsed.get(0) == "0810" and rc == "00"
except (OSError, ValueError) as e:
log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0))
except Exception as e:
log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0, stan=stan))
return False
......@@ -400,17 +337,19 @@ def send_tx(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounter",
send_ts = ts()
t0 = time.perf_counter()
try:
print(f"DEBUG: sending transaction for inv_id={inv_id} with stan={stan} at {send_ts} {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')}")
send_frame(sock, payload, use_etx=use_etx)
resp = recv_frame(sock, expect_etx=use_etx)
print(f"DEBUG: received transaction for inv_id={inv_id} with stan={stan} at {send_ts} {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')}")
dt_ms = (time.perf_counter() - t0) * 1000.0
parsed = parse_until_39(resp)
rc = parsed.get(39)
ok = parsed.get(0) == "0210" and rc == "00"
log_line(format_ok_line(kind="TX", inv_id=inv_id, rc39=rc, send_ts=send_ts, t0=t0))
log_line(format_ok_line(kind="TX", inv_id=inv_id, rc39=rc, send_ts=send_ts, t0=t0, stan=stan))
return TxResult(inv_id=inv_id, ok=ok, rc39=rc, duration_ms=dt_ms)
except (OSError, ValueError) as e:
dt_ms = (time.perf_counter() - t0) * 1000.0
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0))
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0, stan=stan))
return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=dt_ms, error=type(e).__name__)
......@@ -566,13 +505,13 @@ def build_0200_transaction(stan: str, *, bit61: str) -> bytes:
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(description="Minimal ISO8583 sender")
p.add_argument("--host", required=True)
p.add_argument("--port", required=True, type=int)
p.add_argument("--host", default="127.0.0.1", help="Server host (default: 127.0.0.1)")
p.add_argument("--port", default=8585, type=int, help="Server port (default: 8585)")
p.add_argument("--inv-id", dest="bit61", default=None, help="invoice profile string for bit 61")
p.add_argument(
"--inv-file",
default=None,
help="path to invoices.txt (one invoice per line; blank/# ignored)",
default="demo.txt",
help="path to invoices.txt (one invoice per line; blank/# ignored) (default: demo.txt)",
)
p.add_argument(
"--limit",
......@@ -589,8 +528,43 @@ def main(argv: Optional[List[str]] = None) -> int:
p.add_argument("--timeout", type=float, default=25.0)
p.add_argument("--no-etx", action="store_true", help="length prefix only (no trailing ETX)")
p.add_argument("--no-signon", action="store_true", help="skip initial 0800/301")
p.add_argument(
"--network-only",
action="store_true",
help="Only send a network management (0800/301) message and exit."
)
p.add_argument(
"--network-count",
type=int,
default=1,
help="Number of network management (sign-on) messages to send if --network-only is set (default: 1)"
)
args = p.parse_args(argv)
if args.network_only:
use_etx = not args.no_etx
results = []
count = args.network_count
with socket.create_connection((args.host, args.port), timeout=args.timeout) as s:
stan_counter = StanCounter()
threads = []
results = [None] * count
def send_one(idx):
ok = send_signon(s, use_etx=use_etx, stan_counter=stan_counter)
print(f"Network management test {idx+1}/{count} (0800/301) sent, response OK: {ok}")
results[idx] = ok
for i in range(count):
t = threading.Thread(target=send_one, args=(i,))
threads.append(t)
t.start()
for t in threads:
t.join()
return 0 if all(results) else 1
invoices: List[str] = []
if args.inv_file:
invoices = load_invoices_from_file(args.inv_file)
......@@ -630,17 +604,54 @@ def main(argv: Optional[List[str]] = None) -> int:
results: List[TxResult] = []
any_fail = False
try:
with socket.create_connection((args.host, args.port), timeout=args.timeout) as s:
dispatcher = StanDispatcher(s, use_etx=use_etx, timeout_s=args.timeout)
# Shared state for receipt thread
pending: Dict[str, _Pending] = {}
pending_lock = threading.Lock()
closed = threading.Event()
stan_counter = StanCounter()
send_lock = threading.Lock()
# Start receipt thread
receipt_thread = ReceiptThread(s, use_etx, pending, pending_lock, closed)
receipt_thread.start()
# Minimal sign-on using the reduced function
def send_signon_stan():
return send_signon(s, use_etx=use_etx, stan_counter=stan_counter)
def send_tx_stan(inv_id: str) -> TxResult:
stan = stan_counter.next()
payload = build_0200_transaction(stan, bit61=inv_id)
send_ts = ts()
t0 = time.perf_counter()
p = _Pending(kind="TX", inv_id=inv_id, stan=stan, send_ts=send_ts, t0=t0, event=threading.Event())
with pending_lock:
pending[stan] = p
log_send("TX", inv_id, stan, send_ts)
try:
with send_lock:
send_frame(s, payload, use_etx=use_etx)
except (OSError, ValueError) as e:
with pending_lock:
pending.pop(stan, None)
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0, stan=stan))
return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__)
if not p.event.wait(args.timeout):
with pending_lock:
pending.pop(stan, None)
log_line(format_err_line(kind="TX", inv_id=inv_id, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0, stan=stan))
return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error="timeout")
return p.result or TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=p.error or "unknown")
if do_signon:
ok_signon = dispatcher.signon()
ok_signon = send_signon_stan()
if not ok_signon:
any_fail = True
# Still continue to print summary/run end.
# Prepare result slots to preserve invoice ordering in summary.
results = [
TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error="not_processed")
for inv_id in invoices
......@@ -651,7 +662,7 @@ def main(argv: Optional[List[str]] = None) -> int:
def run_one(idx: int, inv_id: str) -> None:
nonlocal any_fail
with sem:
r = dispatcher.tx(inv_id)
r = send_tx_stan(inv_id)
results[idx] = r
if not r.ok:
any_fail = True
......@@ -664,10 +675,11 @@ def main(argv: Optional[List[str]] = None) -> int:
for t in threads:
t.join()
dispatcher.close()
closed.set()
receipt_thread.join(timeout=2)
except (OSError, ValueError) as e:
log_line(format_err_line(kind="CONNECT", inv_id=None, exc=e, send_ts=ts(), t0=time.perf_counter()))
# Mark all as failed if we never managed to connect.
if not results:
results = [TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__) for inv_id in invoices]
any_fail = True
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!