class Connection:
"""Sync RFC Connection facade binding a Transport to a Session (TRANS-04/05/06).
Construct with a Transport, then drive the handshake via ``_handshake`` (the
public ``connect`` factory does this for you). Once READY, ``ping`` /
``get_connection_attributes`` are available; ``close`` is safe in any state.
"""
def __init__(
self,
transport: Transport,
*,
strict_params: bool = False,
metadata_cache: MetadataCache | None = None,
metadata_cache_key: str | None = None,
) -> None:
self._transport = transport
# Unknown-parameter policy (issue #24). Default False mirrors what callers
# porting from pyrfc expect; set True to have call() reject an argument the
# function interface does not declare.
self._strict_params = strict_params
self._dropped_params_seen: set[tuple[str, tuple[str, ...]]] = set()
self._session = Session()
self._lock = threading.Lock()
# A descriptor describes the system, not this socket, so the cache can be
# shared: a pool passes one in and its connections stop each paying for
# the same interfaces. Falls back to a private cache when none is given.
self._cache = metadata_cache if metadata_cache is not None else MetadataCache()
# Used in place of sys_id when the system sends none. A pool supplies one
# shared value, since its connections were opened from identical
# parameters and therefore reach the same system by construction.
self._anon_cache_key: str | None = metadata_cache_key
self._metrics = ConnectionMetrics()
self._struct_desc_cache: dict[str, TypeDesc] = {} # tabname → TypeDesc (META-04)
self._snc_mode: bool = False
# 16 bytes proposed by this client in the LOGON's 0x0514 and echoed in the
# reply, so a caller can correlate the session.
self._ws_session_token: bytes = b""
# Server-reported duration of the most recent call (tag 0x0667, seconds).
# The async core has its own; this one serves the wRFC and SNC paths,
# which do not delegate.
self._last_server_duration_s: float | None = None
self._ws_auth: dict[str, Any] | None = None # stored by _ws_begin for deferred LOGON
# Async delegation (set by connect() for classic TCP paths, D-07).
# None for SNC/wRFC paths which keep the existing sync transport code unchanged.
self._async_conn: AsyncConnection | None = None
self._loop_thread: _LoopThread | None = None
@classmethod
def _from_async(
cls,
async_conn: AsyncConnection,
loop_thread: _LoopThread,
) -> Connection:
"""Create a Connection in async-delegation mode for the classic TCP path (D-07).
The resulting Connection holds an AsyncConnection + _LoopThread and delegates
every public method (call/ping/close/get_connection_attributes/sys_id) to the
async core. SNC/wRFC paths are NOT affected — they use the regular __init__.
"""
inst = object.__new__(cls)
inst._transport = async_conn._transport # type: ignore[assignment]
inst._session = async_conn._session
inst._lock = threading.Lock()
inst._cache = async_conn._cache
inst._struct_desc_cache = async_conn._struct_desc_cache
inst._snc_mode = False
inst._ws_auth = None
inst._async_conn = async_conn
inst._loop_thread = loop_thread
inst._strict_params = async_conn._strict_params
inst._dropped_params_seen = async_conn._dropped_params_seen
return inst
# ------------------------------------------------------------------ #
# Handshake
# ------------------------------------------------------------------ #
def _ws_begin(
self,
*,
client: str,
user: str,
passwd: str,
lang: str = _DEFAULT_LANG,
sysnr: str = "00",
) -> None:
"""Store wRFC auth params and advance to WS_PENDING; no LOGON frame sent.
The RFC LOGON is deferred to the first call() (Track 2 lazy-LOGON design).
_call_bootstrap() sends the combined LOGON+RFC_GET_FUNCTION_INTERFACE frame
when the session is still WS_PENDING. Never logs credentials (T-07-CRED).
"""
try:
peer = self._transport._sock.getpeername()
local = self._transport._sock.getsockname()
local_ip = local[0]
local_port = local[1]
server_host = peer[0]
server_port = peer[1]
except Exception:
local_ip = "127.0.0.1"
local_port = 0
server_host = "127.0.0.1"
server_port = 443
self._ws_auth = {
"user": user,
"passwd": passwd,
"client": client,
"lang": lang,
"local_ip": local_ip,
"local_port": local_port,
"server_host": server_host,
"server_port": server_port,
"sysnr": sysnr,
}
self._session.begin_ws_session()
def _ws_handshake(
self,
*,
client: str,
user: str,
passwd: str,
lang: str = _DEFAULT_LANG,
sysnr: str = "00",
) -> None:
"""Deferred wRFC LOGON setup: store auth, advance to WS_PENDING, wait for first call.
wRFC connect defers the RFC LOGON to the first call(): the LOGON frame names
the function to run in 0x0102, so there is nothing to send until a caller
says which function that is. The frame carries no separate call body -- see
_build_ws_logon_message for the shape and the evidence behind it.
Never logs credentials (T-07-CRED).
"""
try:
peer = self._transport._sock.getpeername()
local = self._transport._sock.getsockname()
local_ip = local[0]
local_port = local[1]
server_host = peer[0]
server_port = peer[1]
except Exception:
local_ip = "127.0.0.1"
local_port = 0
server_host = "127.0.0.1"
server_port = 443
# Store auth so _call_bootstrap (WS_PENDING / Track 2 path) can build the LOGON.
self._ws_auth = {
"user": user,
"passwd": passwd,
"client": client,
"lang": lang,
"local_ip": local_ip,
"local_port": local_port,
"server_host": server_host,
"server_port": server_port,
"sysnr": sysnr,
}
# DISCONNECTED → WS_PENDING; the LOGON frame is deferred to the first call().
self._session.begin_ws_session()
def _handshake(
self,
*,
client: str,
user: str | None,
passwd: str | None,
ashost: str = "0.0.0.0",
sysnr: int = 0,
lang: str = _DEFAULT_LANG,
) -> None:
"""Drive the NI/GW/logon handshake to READY (or raise on failure).
The Session emits the NI-version request; for GW-connect, GW-info,
GW-done, and logon legs the facade supplies the request bytes (the pure
state machine does not own credential/handle framing). We loop, feeding
each server frame and sending the facade-supplied frames, until READY.
"""
# wRFC path: bypass NI/GW entirely; use RFC app-layer TLVs over WebSocket.
try:
from saprfclib.ws import WsTransport
if isinstance(self._transport, WsTransport):
if user is None or passwd is None:
# wRFC authenticates over HTTP on the WebSocket upgrade, so an
# anonymous attempt has nowhere to go — the credentials are not
# carried in the RFC logon frame at all.
raise ValueError(
"WebSocket RFC requires a user and password: the credentials "
"are sent on the HTTP upgrade, so there is no anonymous form "
"of this connection"
)
self._ws_handshake(
client=client,
user=user,
passwd=passwd,
lang=lang,
sysnr=f"{sysnr:02d}",
)
return
except ImportError:
pass
try:
local_ip: str = self._transport._sock.getsockname()[0]
except AttributeError:
# SncTransport has no _sock directly — proxy through inner.
try:
local_ip = self._transport._inner._sock.getsockname()[0] # type: ignore[attr-defined]
except Exception:
local_ip = "127.0.0.1"
except Exception:
local_ip = "127.0.0.1"
if self._session.state is SessionState.DISCONNECTED:
# Standard path: begin NI exchange; loop below receives NI response.
self._transport.send_message(self._session.start(local_ip=local_ip))
elif self._session.state is SessionState.NI_VERSIONED:
# SNC path: NI exchange already completed on the plain inner channel;
# GW connect is the first frame needed (still plain — SNC activates
# after GW_DONE; see activate_snc() call in the loop below).
for req in self._build_leg_requests(
SessionState.CONNECTED,
client=client,
user=user,
passwd=passwd,
ashost=ashost,
sysnr=sysnr,
local_ip=local_ip,
lang=lang,
):
self._transport.send_message(req)
while self._session.state is not SessionState.READY:
resp = self._transport.recv_message()
prev_state = self._session.state
out = self._session.feed(resp)
if out:
self._transport.send_message(out)
else:
# SNC: after GW_DONE server response (prev=GW_CONNECTED) run
# the GSS handshake so the RFC logon goes over the encrypted
# channel. Wire-capture confirmed: GW_INFO+GW_DONE go plain;
# SNC FR_INIT/FR_ACCEPT happen inside 0x06CB GW frames AFTER
# GW_DONE (not between GW_CONNECT and GW_INFO, as first assumed).
if prev_state is SessionState.GW_CONNECTED:
if hasattr(self._transport, "activate_snc"):
self._transport.activate_snc(self._session.handle)
for req in self._build_leg_requests(
prev_state,
client=client,
user=user,
passwd=passwd,
ashost=ashost,
sysnr=sysnr,
local_ip=local_ip,
lang=lang,
):
self._transport.send_message(req)
def _build_leg_requests(
self,
prev_state: SessionState,
*,
client: str,
user: str | None,
passwd: str | None,
ashost: str,
sysnr: int,
local_ip: str,
lang: str = _DEFAULT_LANG,
) -> list[bytes]:
"""Return the facade-owned frame(s) for the leg just advanced past.
NI_VERSIONED → GW_CONNECTED: sends GW_INFO then GW_DONE_CLIENT as two
separate frames (GW_INFO has no server response, so both are sent in the
same iteration before the next recv).
"""
handle = self._session.handle or b"00000000"
match prev_state:
case SessionState.CONNECTED:
return [self._build_gw_connect_request(ashost, sysnr, snc=self._snc_mode)]
case SessionState.NI_VERSIONED:
return [
self._build_gw_info(handle, ashost, snc=self._snc_mode),
self._build_gw_done_client(handle, snc=self._snc_mode),
]
case SessionState.GW_CONNECTED:
tlv = self._build_logon_request(
client=client, user=user, passwd=passwd, local_ip=local_ip, lang=lang
)
if self._snc_mode:
# SNC: encrypt only the RFC application data (COM_HEAD + TLV).
# Outer GW-SNC header (80B) is added by SncTransport._build_gw_snc_frame.
# protocol analysis STIntSend/the SNC output path: arg4 (plain data) = COM_HEAD + TLV — no GW header.
return [_COM_HEAD + tlv]
return [self._build_logon_frame(handle, tlv)]
case _:
return []
# ------------------------------------------------------------------ #
# GW frame builders (facade-owned; Session does not synthesize these)
# ------------------------------------------------------------------ #
@staticmethod
def _build_gw_connect_request(ashost: str, sysnr: int, *, snc: bool = False) -> bytes:
"""Build the 453-byte GW_CONNECT_REQUEST payload (PKT 8 capture).
Confirmed from the GW_CONNECT frame builder.
Fields confirmed by analysis:
[0:2] type = 0x0601
[2:4] version = 0x0200
[4:8] flags = 0xFFFF0000 [4:6]=0xffff, [6:8]=0 (memset)
[10] 0x01 plain / 0x21 SNC (bit 0x20 marks SNC)
[16] 0xC0
[21] 0x04 (a standard client; the registration ACK comes back 0x06)
[22] 0x00
[40:48] " " no handle outbound — the gateway assigns one
[48:56] "NWRFC " remote partner LU name, 8 bytes, net/ASCII
[73] 0x01
[76:78] 0x0000
[78:80] 0xffff (the ACK flips this to 0x0004)
Every fixed byte above is read off the committed capture
tests/golden/framing/server_registration_request.bin.
Remaining bytes: wire-captured from PKT 8 (golden fixture validated).
"""
payload = bytearray(453)
struct.pack_into(">H", payload, 0, _GW_TYPE_CONNECT)
struct.pack_into(">H", payload, 2, _GW_VERSION)
struct.pack_into(">I", payload, 4, _GW_FLAGS)
payload[8:28] = (
b"\x00\x00\x01\x00\x00\x00\x00\x00\xc0\x00\x00\x00\x00\x04\x00\x00\x00\x00\x01\x75"
)
if snc:
payload[10] |= 0x20 # bit 0x20 = SNC capability
payload[28:36] = b"\x00\x00\x05\x00\x00\x00\x00\x00"
payload[40:48] = b" " # no handle in outbound request
payload[48:56] = b"NWRFC " # remote LU name = RFC gateway partner
payload[56:64] = ashost[:8].ljust(8).encode("ascii") # IP prefix
# Two-digit field: "sapdp" + NN + one space is exactly 8 bytes. A value
# above 99 used to make it 9 and grow the whole frame by a byte.
payload[64:72] = f"sapdp{_validate_sysnr(sysnr):02d} ".encode("ascii")
payload[72:80] = (
b"\x49\x01\x00\x00\x00\x00\xff\xff" # [73]=1, [76:78]=0, [78:80]=0xffff (confirmed)
)
payload[80:85] = b"NWRFC"
payload[85:112] = b" " * 27
payload[112:114] = b"\x01\x01"
payload[114:118] = b"CPIC"
# CPIC session ID (32-byte ASCII hex, session-specific)
payload[122:154] = os.urandom(16).hex().upper().encode("ascii")
payload[156:172] = b"\x00\x01\xff\xff\xff\xfe\xff\xff\xff\xfe\x02\x00\x00\x00\x00\x00"
# Server IP null-terminated at payload[185]
ip_b = ashost.encode("ascii") + b"\x00"
payload[185 : 185 + min(len(ip_b), 16)] = ip_b[:16]
# Client hostname null-terminated at payload[329]
try:
hn = _socket_module.gethostname().encode("ascii", "replace") + b"\x00"
except Exception:
hn = b"saprfclib\x00"
payload[329 : 329 + min(len(hn), 16)] = hn[:16]
# Service null-terminated at payload[389]
svc = f"sapdp{sysnr:02d}\x00".encode("ascii")
payload[389 : 389 + min(len(svc), 8)] = svc[:8]
return bytes(payload)
@staticmethod
def _build_gw_info(handle: bytes, ashost: str, *, snc: bool = False) -> bytes:
"""Build the 224-byte GW_INFO payload (PKT 10 capture; no server response).
Confirmed from the GW_INFO frame builder.
Fields confirmed by analysis:
[0:2] type = 0x060F
[4:8] flags = 0xFFFF0000 ([4:6] = 0xffff, [6:8] = 0x0000)
[27] 0x90
[30] 0x04
[40:48] handle the 8-byte ASCII handle the gateway assigned
[76:80] 0xFFFF0004 plain / 0xFFFF0009 with SNC
No committed capture holds a GW_INFO frame; what stands behind these bytes
is that a live gateway completes the handshake when they are sent.
Total size 0xe0=224 bytes: confirmed from the gateway send path(..., 0xe0) in the GW_INFO builder.
payload[8:12], [24:28], [28:32]: wire-captured from PKT 10 golden fixture.
``snc=True`` selects _GW_CLIENT_TAIL_SNC (live pyrfc SNC capture D-24).
"""
payload = bytearray(224)
struct.pack_into(">H", payload, 0, _GW_TYPE_INFO)
struct.pack_into(">H", payload, 2, _GW_VERSION)
struct.pack_into(">I", payload, 4, _GW_FLAGS)
payload[8:12] = b"\x00\x00\x01\x00"
payload[24:28] = b"\x00\x00\x00\x90" # [27]=0x90 confirmed (confirmed)
payload[28:32] = b"\x00\x00\x04\x00" # [30]=4 confirmed (confirmed)
payload[40:48] = handle
payload[48:56] = ashost[:8].ljust(8).encode("ascii")
struct.pack_into(">I", payload, 56, len(ashost))
struct.pack_into(">I", payload, 76, _GW_CLIENT_TAIL_SNC if snc else _GW_CLIENT_TAIL)
# Server IP padded with spaces to 112 bytes at payload[80]
ip_b = ashost.encode("ascii")
padded = ip_b + b" " * (112 - len(ip_b))
payload[80:192] = padded[:112]
return bytes(payload)
@staticmethod
def _build_gw_done_client(handle: bytes, *, snc: bool = False) -> bytes:
"""Build the 80-byte GW_DONE_CLIENT payload (golden fixture + confirmed).
Confirmed from the GW_DONE frame builder.
Fields confirmed by analysis:
[0:2] type = 0x0605
[4:8] flags = 0xFFFF0000 ([4:6] = 0xffff, [6:8] = 0x0000)
[30] 0x01
[40:48] handle the 8-byte ASCII handle the gateway assigned
[76:80] 0xFFFF0004 plain / 0xFFFF0009 with SNC
Source: tests/golden/handshake/gw_done_client.bin (and gw_done_server.bin
for the gateway's reply).
Total size 0x50=80 bytes: confirmed from the gateway send path(..., 0x50) in the GW_DONE builder.
``snc=True`` selects _GW_CLIENT_TAIL_SNC (live pyrfc SNC capture D-24).
"""
payload = bytearray(80)
struct.pack_into(">H", payload, 0, _GW_TYPE_DONE)
struct.pack_into(">H", payload, 2, _GW_VERSION)
struct.pack_into(">I", payload, 4, _GW_FLAGS)
payload[28:32] = b"\x00\x00\x01\x00" # [30]=1 confirmed
payload[40:48] = handle
struct.pack_into(">I", payload, 76, _GW_CLIENT_TAIL_SNC if snc else _GW_CLIENT_TAIL)
return bytes(payload)
@staticmethod
def _build_logon_frame(handle: bytes, tlv_body: bytes, *, snc: bool = False) -> bytes:
"""Wrap TLV body in the RFC logon frame: GW header (76B) + RFC marker + COM_HEAD + TLV.
Byte layout confirmed from stfc_connection.pcapng PKT 14 hex dump:
[0:4] 0x06CB 0x0200 type + version (all GW builders set [0]=6, [1]=type_lsb)
[4:8] 0xFFFF0000 flags ([4:6]=0xffff hardcoded, [6:8]=0 from memset)
[24:28] 0x00000008 APPC header version (must be 8 for NW 7.x) — _GW_HDR_APPC_VER
[28:32] 0x0000050C CPIC max message length = 1292 — _GW_HDR_MAX_LEN
[40:48] handle 8-byte ASCII GW handle
[76:80] RFC_MARKER FF FF 00 04 (plain) / FF FF 00 09 (SNC)
[80:92] COM_HEAD EBCDIC "RFC000000000"
[92:] TLV body
"""
gw = bytearray(76)
struct.pack_into(">H", gw, 0, _GW_TYPE_RFC)
struct.pack_into(">H", gw, 2, _GW_VERSION)
struct.pack_into(">I", gw, 4, _GW_FLAGS)
struct.pack_into(">I", gw, 24, _GW_HDR_APPC_VER)
struct.pack_into(">I", gw, 28, _GW_HDR_MAX_LEN)
gw[40:48] = handle
marker = struct.pack(">I", _GW_CLIENT_TAIL_SNC if snc else _GW_CLIENT_TAIL)
return bytes(gw) + marker + _COM_HEAD + tlv_body
@staticmethod
def _build_invoke_frame(handle: bytes, tlv_body: bytes) -> bytes:
"""Wrap TLV body in an RFC invoke frame: GW header (76B) + RFC marker + TLV.
Invoke frames omit COM_HEAD (present only in the logon frame). Confirmed by
comparing stfc_connection_request.bin golden (client invoke request) against
_build_logon_frame: the invoke frame has no EBCDIC COM_HEAD between the RFC
marker and the TLV body.
Wire layout (wire-captured from stfc_connection_request.bin):
[0:4] 0x06CB 0x0200 type + version (same as logon frame)
[4:8] 0xFFFF0000 flags
[24:28] 0x00000008 APPC header version (must be 8 for NW 7.x)
[28:32] 0x0000050C CPIC max message length = 1292 (NW 7.x)
[40:48] handle 8-byte ASCII GW handle
[76:80] RFC_MARKER FF FF 00 04
[80:] TLV body (NO COM_HEAD — invoke frames only)
Omitting GW[24:32] causes immediate 80B 0x06CE rejection from the server
("client with wrong appc header version rejected").
Footer: every invoke frame ends with an 8-byte trailer inside the NI frame:
[0:4] uint32 BE len(tlv_body) | [4:6] 0x0000 | [6:8] 0x8500
Wire-verified in all nine request fixtures; absent from server responses
(responses carry a 0x0667 timing double instead). The length is 32-bit: a
uint16 fits every capture only because every captured body is small, and
overflows for bodies above 64 KB. See _INVOKE_FOOTER_MAGIC.
"""
gw = bytearray(76)
struct.pack_into(">H", gw, 0, _GW_TYPE_RFC)
struct.pack_into(">H", gw, 2, _GW_VERSION)
struct.pack_into(">I", gw, 4, _GW_FLAGS)
struct.pack_into(">I", gw, 24, _GW_HDR_APPC_VER)
struct.pack_into(">I", gw, 28, _GW_HDR_MAX_LEN)
gw[40:48] = handle
footer = struct.pack(">I", len(tlv_body)) + _INVOKE_FOOTER_MAGIC
return bytes(gw) + _RFC_MARKER + tlv_body + footer
def _send_invoke_frame(self, frame: bytes) -> None:
"""Send an RFC invoke frame to the transport.
For SNC, strip the outer 80B GW header (76B header + 4B RFC_MARKER) —
SncTransport._build_gw_snc_frame builds its own GW envelope, so the
encrypted payload must be only TLV+footer (same protocol analysis logic as logon).
For non-SNC, send the full GW-framed bytes unchanged.
This method is the classic/SNC GW path ONLY. The wRFC transport bypasses it
entirely: _call_bootstrap / call() / _call_struct_bootstrap send raw wRFC
frames directly via self._transport.send_message when self._is_ws() is true,
so no WsTransport branch is needed here.
"""
if self._snc_mode:
self._transport.send_message(frame[80:])
else:
self._transport.send_message(frame)
@staticmethod
def _build_logon_request(
*,
client: str,
user: str | None,
passwd: str | None,
seed: int | None = None,
local_ip: str = "127.0.0.1",
program_name: bytes = b"python3",
lang: str = _DEFAULT_LANG,
) -> bytes:
"""Build the RFC logon TLV body in extended wire format (tag+len+val+tag).
Emits the scrambled password record (tag 0x0117) per the RE-confirmed
derivation (Plan 04-01: ``seed(4B) + scramble(password, seed)``). The
plaintext ``passwd`` is scrambled, never emitted plaintext and never
logged (threat T-04-CRED / T-03-CRED2). ``seed`` is injectable so offline
tests are deterministic; production uses a fresh per-call client nonce.
"""
try:
hn = _socket_module.gethostname().encode("ascii", "replace")
except Exception:
hn = b"saprfclib"
session_token = os.urandom(16)
parts = [
_tlv_ext(0x0101, _TLV_CAPS),
_tlv_ext(0x0103, _TLV_VER),
_tlv_ext(0x0106, _TLV_CP),
_tlv_ext(0x0514, session_token),
_tlv_ext(_TAG_CLIENT, client.encode("ascii", "replace")),
]
# No credentials: omit the user and password records rather than sending
# empty ones. An empty password is still a password attempt as far as the
# server is concerned, and repeated attempts against a real account name
# count towards lockout; omitting the fields cannot.
if user is not None:
parts.append(_tlv_ext(_TAG_USER, user.encode("ascii", "replace")))
if passwd is not None:
parts.append(_tlv_ext(_TAG_PASSWORD, _scramble_password(passwd, seed=seed)))
parts += [
# 0x0115 and 0x0011 both carry the logon language in the capture
# (golden logon_request.bin: b"E" on each).
_tlv_ext(0x0115, _encode_logon_language(lang)),
_tlv_ext(0x0501, b"\x01"),
_tlv_ext(0x0007, b"127.0.0.1"),
_tlv_ext(0x0011, _encode_logon_language(lang)),
_tlv_ext(0x0012, _TLV_REL),
_tlv_ext(0x0013, _TLV_REL),
_tlv_ext(0x0008, hn),
_tlv_ext(0x0006, _TLV_PROG),
_tlv_ext(0x0130, program_name),
_tlv_ext(0x0502, b""),
_tlv_ext(0x000B, _TLV_REL),
_tlv_ext(_TAG_FUNCTION, _RFCPING_NAME),
# Terminator: no repeated tag
_TAG_TERMINATOR.to_bytes(2, "big") + b"\x00\x00",
# Trailing call-frame marker. Behavioural evidence only: the gateway
# accepts this frame and answers the ping.
b"\xff\xff\x00\x00\x00\xf8\x00\x00\x85\x00",
]
return b"".join(parts)
# ------------------------------------------------------------------ #
# Public surface
# ------------------------------------------------------------------ #
@property
def metrics(self) -> ConnectionMetrics:
"""Per-connection call counters and latency.
Delegates to the async core for classic TCP connections (D-07), so the
numbers are the same object whichever facade the caller holds.
"""
if self._async_conn is not None:
return self._async_conn.metrics
return self._metrics
@property
def _metadata_cache_key(self) -> str | None:
"""Key this connection's cached descriptors live under; None to not cache.
Normally the system ID, so every connection to the same system shares one
set of descriptors. But the logon response does not always carry one: a
7.52 system answers with no 0x0450/0x0452/0x0453 at all, leaving sys_id
empty. Caching under "" would file every such system in one bucket, and a
process holding connections to two of them would be served the wrong
system's descriptor for a same-named function module — silently, since a
FunctionDesc carries no system of origin.
So an unidentified system falls back to a token unique to this connection.
Repeat calls on the connection still skip the round-trip; nothing is
shared between systems that never identified themselves.
"""
sys_id = self.sys_id
if sys_id is None:
return None # not READY — nothing to key on yet
if sys_id:
return sys_id
if self._anon_cache_key is None:
# NUL prefix: a real SID is 3 alphanumerics, so this cannot collide.
self._anon_cache_key = f"\x00anon-{uuid.uuid4().hex}"
return self._anon_cache_key
@property
def sys_id(self) -> str | None:
"""System ID from the negotiated ConnectionAttributes; None if not READY.
Used by get_function_desc as the cache key ((sys_id, func_name) tuple).
Delegates to async core for classic TCP connections (D-07).
"""
if self._async_conn is not None:
return self._async_conn.sys_id
attrs = self._session.attributes
return attrs.sys_id if attrs is not None else None
def _ensure_ws_session(self) -> None:
"""Complete the deferred wRFC LOGON, if it has not happened yet.
A wRFC connection does the HTTP upgrade in ``connect()`` and defers the
LOGON to the first call, so it sits in WS_PENDING until something needs
the session. That is a reasonable design and a poor one to expose: a
caller who opened a connection and asked to ``ping()`` it got
``operation not allowed in state 'WS_PENDING'``, which describes the
library's internal bookkeeping rather than anything they did wrong, and
offers no way forward.
The LOGON names RFCPING in its own 0x0102 and the server runs it, so
completing the session here is itself the liveness check ``ping()`` was
asking for -- there is no extra round trip.
No-op on any other transport or state, so callers can invoke it
unconditionally.
"""
if not self._is_ws() or self._session.state is not SessionState.WS_PENDING:
return
auth = self._ws_auth or {}
logon_msg, session_token = _build_ws_logon_message(
func_name="RFCPING",
user=auth["user"],
passwd=auth["passwd"],
client=auth["client"],
lang=auth["lang"],
local_ip=auth["local_ip"],
)
self._ws_session_token = session_token
with _fail_closed(self._session, "RFCPING"):
self._transport.send_message(logon_msg)
logon_resp = _join_response_frames(self._transport.recv_message, "LOGON")
attrs_ws = _ws_parse_logon_response(logon_resp)
failure = _ws_logon_failure(logon_resp)
if failure is not None:
if attrs_ws and attrs_ws.sys_id:
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
raise failure
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
def get_connection_attributes(self) -> ConnectionAttributes:
"""Return the negotiated ConnectionAttributes (populated at READY, TRANS-07).
Delegates to async core for classic TCP connections (D-07).
"""
if self._async_conn is not None:
return self._async_conn.get_connection_attributes()
# On wRFC the attributes only exist once the LOGON has run, and the LOGON
# is deferred. Asking for them is a reasonable way to say "establish the
# session", so do that rather than reporting an internal state.
self._ensure_ws_session()
attrs = self._session.attributes
if attrs is None:
raise ValueError("connection is not in READY state")
return attrs
def _is_ws(self) -> bool:
"""True if the transport is a wRFC WebSocket transport (lazy import).
Mirrors the lazy-import guard used at handshake time. wRFC carries the same
invoke TLV stream as classic RFC but without the GW header, so the only
thing this decides is whether to wrap a frame in that header -- the
builders and parsers are shared. Returns False if the optional ws module
is unavailable.
"""
try:
from saprfclib.ws import WsTransport
except ImportError:
return False
return isinstance(self._transport, WsTransport)
def _call_bootstrap(self, func_name: str) -> FunctionDesc:
"""Bootstrap invoke to fetch FunctionDesc via RFC_GET_FUNCTION_INTERFACE (D-21).
Sends the RFC_GET_FUNCTION_INTERFACE TLV using the bootstrap descriptor
(BOOTSTRAP_GET_FUNCTION_INTERFACE) to avoid the chicken-and-egg problem:
we cannot call get_function_desc for RFC_GET_FUNCTION_INTERFACE because
that would require RFC_GET_FUNCTION_INTERFACE's own metadata.
This method is NOT protected by the single-in-flight lock because it is
always called from within call() which already holds the lock. It accesses
the transport directly (below the CPIC state machine).
Parses the PARAMS TABLE from the response using a simplified path that
walks 0x0201/0x0203 pairs to extract the table rows as dicts with the
confirmed 12-column layout (META-01 columns confirmed 2026-06-27).
On wRFC the same request TLV is sent without a GW header; on classic and
SNC it is wrapped in one. Nothing else differs between the two paths.
OSError/EOFError propagate to call()'s CommunicationError wrapper.
"""
# Classic TCP path: delegate to the async core (D-07), as every other
# method on this class does. Without this the bootstrap ran its sync body
# against _SyncToAsyncTransport, whose send/recv are coroutines: the frame
# was never sent and the "response" was a coroutine object, surfacing as
# "TypeError: 'coroutine' object is not subscriptable". That made the
# public metadata.get_function_desc() unusable on any classic connection.
if self._async_conn is not None and self._loop_thread is not None:
return cast(
FunctionDesc,
self._loop_thread.run(self._async_conn._call_bootstrap(func_name)),
)
attrs = self._session.attributes
unicode_mode = attrs.unicode_mode if attrs else True
# Build the bootstrap invoke request (FUNCNAME = func_name, EXPORTING = PARAMS).
# We add PARAMS as an EXPORTING param decl so the server sends it back.
# The bootstrap descriptor knows FUNCNAME; we add PARAMS manually.
bootstrap_params = [
FieldDesc(
name="FUNCNAME",
rfctype=0, # RFCTYPE_CHAR
nuc_length=30,
nuc_offset=0,
uc_length=60,
uc_offset=0,
decimals=0,
unicode_mode=unicode_mode,
direction=RFC_IMPORT,
),
# PARAMS is an EXPORTING TABLE param, declared so the server sends it
# back. Only the declaration goes out -- an EXPORT param carries no
# value from the client -- and the 12-column reply is parsed above.
FieldDesc(
name="PARAMS",
rfctype=5, # RFCTYPE_TABLE
nuc_length=0,
nuc_offset=0,
uc_length=0,
uc_offset=0,
decimals=0,
unicode_mode=unicode_mode,
direction=RFC_EXPORT,
),
]
bootstrap_desc = FunctionDesc(
name="RFC_GET_FUNCTION_INTERFACE",
parameters=bootstrap_params,
)
_ws_pending_path = False
if self._is_ws():
if self._session.state is SessionState.WS_PENDING:
# 2-step lazy LOGON (Track 2):
# Step 1: LOGON, with RFCPING as the function it runs.
# The LOGON frame is built to the shape a server accepts: no
# 0x5001 record, single-byte strings, and the function to run
# named in 0x0102. See _build_ws_logon_message for how that was
# established and what the previous shape got wrong.
_ws_pending_path = True
auth = self._ws_auth or {}
logon_msg, session_token = _build_ws_logon_message(
func_name="RFCPING",
user=auth["user"],
passwd=auth["passwd"],
client=auth["client"],
lang=auth["lang"],
local_ip=auth["local_ip"],
)
self._ws_session_token = session_token
self._transport.send_message(logon_msg)
logon_resp = _join_response_frames(self._transport.recv_message, "LOGON")
# Auth: extract ConnectionAttributes from 0x0450/0x0452/0x0453.
attrs_ws = _ws_parse_logon_response(logon_resp)
# ... and then check whether the call embedded in that LOGON
# actually ran. The auth tags are filled in either way, so a reply
# that authenticated and then failed reads as a clean logon to
# anything that only looks for a sys_id. Sending an invoke into
# that session is what makes the work process take a short dump.
if (logon_failure := _ws_logon_failure(logon_resp)) is not None:
if attrs_ws and attrs_ws.sys_id:
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
raise logon_failure
if close_exc := self._transport.drain_queued_close(): # type: ignore[attr-defined]
if attrs_ws and attrs_ws.sys_id:
# Auth succeeded (0x0450/sys_id present) and the server then
# closed the WebSocket. Complete the attributes so
# get_connection_attributes() still works, and report what
# the close actually said.
#
# This used to raise a hardcoded "163: Error when receiving
# data for an RFC." The value was right and the sourcing was
# not: the server does send E=163, inside the 0x0418
# call-stack breadcrumb, and nothing was reading it. A
# hardcoded constant that happens to match is still a defect
# -- it reports 163 for every failure, including the ones
# that are not 163. _ws_logon_failure now parses the field.
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
raise AbapSystemFailure(
message=(
f"the server authenticated the wRFC LOGON and then closed "
f"the WebSocket: {close_exc}"
)
) from close_exc
raise close_exc
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
# Step 2: INVOKE+RFC_GET_FUNCTION_INTERFACE (now in READY state).
frame = _build_ws_invoke_frame(
"RFC_GET_FUNCTION_INTERFACE",
bootstrap_desc,
{"FUNCNAME": func_name},
)
self._transport.send_message(frame)
try:
response = _join_response_frames(
self._transport.recv_message, "RFC_GET_FUNCTION_INTERFACE"
)
except WebSocketError as ws_exc:
# The server closed the WebSocket instead of answering
# RFC_GET_FUNCTION_INTERFACE. Auth already completed, so this is
# a function-level failure rather than a transport one -- but
# report the close the server actually sent rather than a
# constant standing in for it.
raise AbapSystemFailure(
message=(
f"the server closed the WebSocket without answering "
f"RFC_GET_FUNCTION_INTERFACE: {ws_exc}"
)
) from ws_exc
else:
# Subsequent bootstrap: connection already established, use invoke format.
frame = _build_ws_invoke_frame(
"RFC_GET_FUNCTION_INTERFACE",
bootstrap_desc,
{"FUNCNAME": func_name},
)
self._transport.send_message(frame)
response = _join_response_frames(
self._transport.recv_message, "RFC_GET_FUNCTION_INTERFACE"
)
else:
request_tlv = build_invoke_request(
"RFC_GET_FUNCTION_INTERFACE",
bootstrap_desc,
{"FUNCNAME": func_name},
)
# Wrap TLV in a GW invoke frame (GW header + RFC marker, no COM_HEAD).
# Raw TLV cannot be sent directly — server validates the GW header and
# rejects the frame with "wrong apppc header version" if bare TLV is sent.
handle = self._session.handle or b" "
frame = self._build_invoke_frame(handle, request_tlv)
self._send_invoke_frame(frame)
# A function interface can be large -- 44 parameters already fill 2342
# bytes -- so this reply chunks like any other.
response = _join_response_frames(
self._transport.recv_message, "RFC_GET_FUNCTION_INTERFACE"
)
# Parse the response TLV to extract PARAMS table rows.
# We use a direct walker rather than parse_invoke_response because we need
# to interpret the raw bytes as PARAMS rows without a TypeDesc descriptor.
# A function module that is not remote-enabled answers GFI with a normal ABAP
# exception (FL/046/FU_NOT_FOUND), and an exception reply carries no 0x0420 —
# so the return-code check never fires and we used to hand back an empty
# descriptor instead. Classify before parsing rows, on every path.
raise_for_rfc_error(_strip_gw_header(response))
rows = _parse_gfi_params_rows(response, unicode_mode=unicode_mode)
if not rows and not _metadata_reply_succeeded(response):
# An empty PARAMS table is only worth reporting when the reply did not
# say it succeeded. A function module with no parameters is legal --
# RFC_PING has none -- so warning on the row count alone cried wolf on
# every parameterless function while saying its descriptor was broken.
_logger.warning(
"no parameter rows parsed from the %s metadata response (%d bytes), "
"and the reply carries no success marker; the descriptor will be "
"empty and calls will reject all arguments",
func_name.upper(),
len(response),
)
# An empty PARAMS table is not a failure. RFC_PING takes no parameters, so
# its interface legitimately has no rows, and treating "no rows" as an
# error made every parameterless function uncallable over wRFC while
# reporting something that had not happened.
#
# Whether the fetch failed is a question the reply already answers:
# 0x0417 marks an exception and 0x0420 carries the return code. Ask those
# rather than inferring from the row count.
if _ws_pending_path and not rows:
_tlv_map = Session._parse_tlv(response)
_rc_raw = _tlv_map.get(0x0420) or b""
_rc = struct.unpack(">I", _rc_raw)[0] if len(_rc_raw) == 4 else 0
_is_exception = 0x0417 in _tlv_map
if _is_exception or _rc:
_exc_raw = _tlv_map.get(0x0411) or b""
_exc_name = (
_exc_raw.decode("utf-16-le", errors="replace").rstrip("\x00 ")
if _exc_raw
else ""
)
_err_msg = _decode_error_text(_tlv_map.get(0x0402))
_detail = _exc_name or _err_msg or "RFC_GET_FUNCTION_INTERFACE failed"
raise AbapSystemFailure(message=f"{_rc}: {_detail}" if _rc else _detail)
# Build FunctionDesc from the parsed rows. Track STRUCTURE params needing
# a secondary RFC_GET_STRUCTURE_DEFINITION bootstrap (META-04).
parameters = []
struct_lookups: list[tuple[FieldDesc, str]] = []
for row in rows:
try:
fd = _parse_params_row(row)
parameters.append(fd)
# TABLE params need the row layout just as much as STRUCTURE params
# do: _parse_params_row promotes PARAMCLASS 'T' rows to RFCTYPE_TABLE
# (see metadata._parse_params_row), so gating this lookup on
# STRUCTURE alone would leave every TABLES param with type_desc=None
# and make build_invoke_request refuse to encode its rows.
if fd.rfctype in (RFCTYPE_STRUCTURE, RFCTYPE_TABLE):
tabname = row.get("TABNAME", "")
if tabname:
struct_lookups.append((fd, tabname))
except ValueError as exc:
# Exception rows are expected here and are not parameters.
if is_exception_row(row):
continue
# A parameter we cannot parse is a real problem: it will be missing
# from the descriptor, so build_invoke_request will reject any value
# the caller passes for it and the server will never return it.
# Never drop one without saying so (T-03-META: the row is untrusted,
# so keep parsing the rest rather than aborting the whole call).
_logger.warning(
"ignoring unparseable metadata row for %s parameter %r: %s",
func_name.upper(),
row.get("PARAMETER", "<unnamed>"),
exc,
)
continue
# Secondary bootstrap: fetch TypeDesc for each STRUCTURE param's row layout.
# Uses _call_struct_bootstrap which calls RFC_GET_STRUCTURE_DEFINITION.
# Results cached in _struct_desc_cache keyed by TABNAME (META-04).
for fd, tabname in struct_lookups:
if tabname not in self._struct_desc_cache:
try:
self._struct_desc_cache[tabname] = self._call_struct_bootstrap(tabname)
except Exception as exc:
# Leaving type_desc=None makes encode/decode fail later with no
# hint as to which lookup went wrong, so record it here.
_logger.warning(
"could not fetch the layout of DDIC type %r; parameter %r "
"cannot be encoded or decoded: %s",
tabname,
fd.name,
exc,
)
td = self._struct_desc_cache.get(tabname)
if td is not None:
fd.type_desc = td
return FunctionDesc(name=func_name.upper(), parameters=parameters)
def _call_struct_bootstrap(self, tabname: str) -> TypeDesc:
"""Fetch RFCTEST field layout via RFC_GET_STRUCTURE_DEFINITION (META-04).
Secondary bootstrap called from _call_bootstrap when GFI returns STRUCTURE
params (EXID='u'). Uses a hardcoded FunctionDesc to avoid the chicken-and-egg
problem. Not protected by the in-flight lock (always called from _call_bootstrap
which is called from call() which already holds the lock).
RFC_GET_STRUCTURE_DEFINITION interface (confirmed 2026-06-29 via live GFI):
TABNAME (I, CHAR C30 = 60B UC) — structure name to look up
FIELDS (T, STRUCTURE, 140B/row) — DFIES rows with field layout
FIELDS rows are parsed by _parse_dfies_rows (140B wire-confirmed layout).
UC offsets are computed by _build_type_desc_from_dfies (alignment rules
verified against RFCTEST golden stfc_structure_request.bin).
OSError/EOFError propagate to call()'s CommunicationError wrapper.
"""
attrs = self._session.attributes
unicode_mode = attrs.unicode_mode if attrs else True
# Hardcoded FunctionDesc for RFC_GET_STRUCTURE_DEFINITION:
# TABNAME=IMPORT CHAR(30), FIELDS=EXPORT TABLE (get 0x0205 decl so server returns it).
rsd_desc = FunctionDesc(
name="RFC_GET_STRUCTURE_DEFINITION",
parameters=[
FieldDesc(
name="TABNAME",
rfctype=RFCTYPE_CHAR,
nuc_length=30,
nuc_offset=0,
uc_length=60,
uc_offset=0,
decimals=0,
unicode_mode=unicode_mode,
direction=RFC_IMPORT,
),
FieldDesc(
name="FIELDS",
rfctype=RFCTYPE_TABLE,
nuc_length=0,
nuc_offset=0,
uc_length=0,
uc_offset=0,
decimals=0,
unicode_mode=unicode_mode,
direction=RFC_EXPORT,
),
],
)
if self._is_ws():
# wRFC: route STRUCTURE lookups through the raw-TLV invoke builder so
# STRUCTURE params over wRFC are attempted, not silently dropped.
frame = _build_ws_invoke_frame(
"RFC_GET_STRUCTURE_DEFINITION", rsd_desc, {"TABNAME": tabname}
)
self._transport.send_message(frame)
else:
request_tlv = build_invoke_request(
"RFC_GET_STRUCTURE_DEFINITION",
rsd_desc,
{"TABNAME": tabname},
)
handle = self._session.handle or b" "
frame = self._build_invoke_frame(handle, request_tlv)
self._send_invoke_frame(frame)
# A DDIC structure definition is a table of field rows and can exceed one
# frame for a wide structure.
response = _join_response_frames(
self._transport.recv_message, "RFC_GET_STRUCTURE_DEFINITION"
)
raise_for_rfc_error(_strip_gw_header(response))
dfies_rows = _parse_dfies_rows(response)
return _build_type_desc_from_dfies(tabname, dfies_rows)
@staticmethod
def _rfcping_request_tlv() -> bytes:
"""Build the RFCPING invoke TLV body.
RFCPING is an ordinary zero-parameter function call, not a special frame —
the logon TLV itself ends with one (tag 0x0102, see handshake.md). Building
it through ``build_invoke_request`` keeps it on the capture-confirmed invoke
path instead of hand-rolling a second TLV writer.
"""
return build_invoke_request("RFCPING", FunctionDesc(name="RFCPING", parameters=[]), {})
def ping(self) -> bool:
"""Issue an RFC-level RFCPING and report liveness (TRANS-05).
Under the single-in-flight lock: require READY, flip to IN_CALL, send the
RFCPING invoke frame, read the response, and check the return-code TLV
(0x0420 == 0). Always restores READY in ``finally`` (TRANS-04).
Delegates to the async core via _LoopThread for classic TCP connections (D-07).
The probe is a fully framed invoke — GW header, RFC marker, TLV body and
footer — exactly like any other call. A bare TLV body reaches the gateway as
a malformed frame and draws a plain-text error back instead of a response.
"""
if self._async_conn is not None and self._loop_thread is not None:
return bool(self._loop_thread.run(self._async_conn.ping()))
with self._lock:
# On wRFC this both establishes the session and answers the question:
# the LOGON names RFCPING and the server runs it.
if self._is_ws() and self._session.state is SessionState.WS_PENDING:
self._ensure_ws_session()
return True
self._session._require_state(SessionState.READY)
self._session.mark_in_call()
try:
request_tlv = self._rfcping_request_tlv()
if self._is_ws():
frame = _build_ws_invoke_frame(
"RFCPING", FunctionDesc(name="RFCPING", parameters=[]), {}
)
self._transport.send_message(frame)
else:
handle = self._session.handle or b" "
self._send_invoke_frame(self._build_invoke_frame(handle, request_tlv))
with _fail_closed(self._session, "RFCPING"):
resp = _join_response_frames(self._transport.recv_message, "RFCPING")
return self._rfcping_ok(resp)
finally:
# Guarded: a failed ping leaves the session BROKEN, and mark_ready
# refuses any state but IN_CALL. Without the guard the finally
# would raise over the top of the real error and hide it.
if self._session.state is SessionState.IN_CALL:
self._session.mark_ready()
@staticmethod
def _rfcping_ok(resp: bytes) -> bool:
"""Parse the RFCPING response; True iff the return-code TLV 0x0420 == 0.
Walks the same wire dialect every other reader in the tree handles — a
live response is a GW frame, its records use the extended-length form for
payloads >= 0xFFFF, and each record is followed by a repeated close tag
(session._parse_tlv, invoke._extract_name_value_pairs,
_parse_gfi_params_rows all do this). Skipping the close tag is not
optional: without it the walk desynchronises by two bytes after the first
record and every subsequent tag and length is read out of garbage, which
surfaces as a bogus "length exceeds remaining payload" on any response
that does not happen to put 0x0420 first.
"""
resp = _strip_gw_header(resp)
pos = 0
n = len(resp)
while pos + 4 <= n:
tag = int.from_bytes(resp[pos : pos + 2], "big")
length = int.from_bytes(resp[pos + 2 : pos + 4], "big")
pos += 4
if tag == _TAG_TERMINATOR:
break
if length == 0xFFFF:
# Extended form: 4B BE length follows the 0xFFFF marker.
if pos + 4 > n:
raise ValueError(
f"malformed RFCPING response: tag 0x{tag:04x} extended form "
f"but buffer too short for ext_len ({n - pos} bytes remain)"
)
length = int.from_bytes(resp[pos : pos + 4], "big")
pos += 4
end = pos + length
if end > n:
raise ValueError(
f"malformed RFCPING response: tag 0x{tag:04x} length {length} "
f"exceeds remaining payload ({n - pos} bytes)"
)
if tag == _TAG_RETURN_CODE:
if length != 4:
raise ValueError(f"RFCPING return code TLV has length {length}, expected 4")
return int.from_bytes(resp[pos:end], "big") == 0
pos = end
# Skip the optional repeated-tag suffix used in extended TLV format.
if pos + 2 <= n and int.from_bytes(resp[pos : pos + 2], "big") == tag:
pos += 2
raise ValueError("RFCPING response missing return-code TLV 0x0420")
def _ws_classic_fallback(
self,
func_name: str,
desc: FunctionDesc,
params: dict[str, Any],
attrs_ws: ConnectionAttributes,
) -> dict[str, Any]:
"""Classic RFC fallback when the wRFC session cannot be completed.
The wRFC LOGON shape is settled (issue #14): no 0x5001 record, single-byte
strings, the function to run named in 0x0102. A server that accepts it
needs no fallback. This path is for the ones that do not -- the LOGON reply
carries an exception instead of a result, or the server closes the
WebSocket after authenticating. The auth tags are filled in either way, so
such a reply reads as a clean logon to anything that only checks for a
sys_id; reading the result is what catches it.
The call is then re-run over a classic TCP RFC connection derived from the
LOGON response, transparently to the caller:
1. Extract partner_host (0x0453) and sys_number (0x0452) from attrs_ws.
2. Open a classic TCP connection to partner_host:3300+sysnr.
3. Drive the full NI/GW/logon handshake to READY.
4. Execute func_name via the classic RFC invoke path.
5. Permanently replace self._transport + self._session with the classic
ones so future calls on this Connection continue to work; _is_ws()
will return False after this method returns.
Never logs credentials (T-07-CRED).
"""
auth = self._ws_auth or {}
partner_host = (attrs_ws.partner_host or "").strip() or auth.get("server_host", "")
sys_number = (attrs_ws.sys_number or "").strip() or auth.get("sysnr", "00")
sysnr_int = int(sys_number) if sys_number.isdigit() else 0
# Close dead wRFC transport (best-effort).
try:
self._transport.close()
except Exception:
pass
# Classic TCP connection + full NI/GW/logon handshake.
tcp = connect_tcp(partner_host, 3300 + sysnr_int)
classic = Connection(tcp)
classic._handshake(
client=auth.get("client", ""),
user=auth.get("user", ""),
passwd=auth.get("passwd", ""),
ashost=partner_host,
sysnr=sysnr_int,
)
# Pre-populate cache with the builtin desc to skip GFI round-trip.
classic_key = classic._metadata_cache_key
if classic_key:
classic._cache.put(classic_key, desc)
result = classic.call(func_name, **params)
# Permanently downgrade: steal classic transport+session+cache.
# After this self._is_ws() == False; subsequent calls use classic RFC.
self._transport = classic._transport
self._session = classic._session
self._cache = classic._cache
self._ws_auth = None
return result
def _ws_direct_logon_call(
self, func_name: str, desc: FunctionDesc, params: dict[str, Any]
) -> dict[str, Any]:
"""WS_PENDING: LOGON+RFCPING then INVOKE+func_name (two round-trips).
Two-step protocol:
Step 1: LOGON frame naming RFCPING as the function to run -- authenticates
and establishes the wRFC session. RFCPING takes no parameters, so
the frame carries declarations only.
Step 2: INVOKE frame with func_name + params. A wRFC invoke is byte-for-byte
a classic invoke TLV stream sent without a GW header, so this is
build_invoke_request's output unchanged.
Transitions WS_PENDING → READY (via complete_ws_first_call) between step 1 and 2.
Caller must hold self._lock. Called only when state is WS_PENDING.
If the server refuses the LOGON or closes the WebSocket after it, the call
is re-run over classic TCP RFC instead (transparent to the caller).
"""
auth = self._ws_auth or {}
# Step 1: LOGON, with RFCPING as the function it runs.
logon_msg, session_token = _build_ws_logon_message(
func_name="RFCPING",
user=auth["user"],
passwd=auth["passwd"],
client=auth["client"],
lang=auth["lang"],
local_ip=auth["local_ip"],
)
self._ws_session_token = session_token
try:
self._transport.send_message(logon_msg)
logon_resp = _join_response_frames(self._transport.recv_message, "LOGON")
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
# Extract auth (0x0450 → sys_id, etc.) — raises ValueError on auth failure.
attrs_ws = _ws_parse_logon_response(logon_resp)
if _ws_logon_failure(logon_resp) is not None:
# The reply authenticated and reported that the call embedded in the
# LOGON failed. Going on to send the invoke anyway is what makes the
# work process take a short dump -- so this path used to provoke a
# RABAX on the server, catch the resulting WebSocket close, and only
# then fall back. Reading the failure here skips the doomed frame
# entirely: same outcome for the caller, one fewer entry in ST22 per
# connection attempt.
return self._ws_classic_fallback(func_name, desc, params, attrs_ws)
if self._transport.drain_queued_close(): # type: ignore[attr-defined]
# Auth passed and the server then closed the WebSocket. The frame shape
# is the accepted one, so this is the server declining to carry the
# session rather than a malformed request. Fall back to classic TCP.
return self._ws_classic_fallback(func_name, desc, params, attrs_ws)
self._session.complete_ws_first_call(attributes=attrs_ws, codepage="4103")
# Cache the descriptor now that the attributes are on the session.
ws_key = self._metadata_cache_key
if ws_key:
self._cache.put(ws_key, desc)
# Step 2: the invoke. No session key -- a reference client's invoke carries
# no 0x0136, and the server never issued one to echo.
invoke_msg = _build_ws_invoke_frame(func_name, desc, params)
try:
self._transport.send_message(invoke_msg)
invoke_resp = _join_response_frames(self._transport.recv_message, func_name)
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
except WebSocketError:
# WS close arrived after step 1 (different TCP segment than LOGON response).
# Same root cause; fall back to classic RFC.
return self._ws_classic_fallback(func_name, desc, params, attrs_ws)
# A wRFC response is a classic response TLV stream without the GW header,
# so the classic parser reads it. Confirmed against reference captures:
# the metadata reply, the logon reply and a UCON rejection all parse, the
# last one surfacing as the ABAP error it is.
result = parse_invoke_response(invoke_resp, desc)
return _convert_date_time_fields(result, desc)
def call(self, func_name: str, **params: object) -> dict[str, Any]:
"""Invoke an RFC function module and return a native-typed dict (CLIENT-01..07).
Protocol:
1. Acquire the single-in-flight lock (TRANS-04).
2. Require READY state; flip to IN_CALL.
3. Resolve sys_id from session attributes for the cache key.
4. Fetch the FunctionDesc via the metadata cache or bootstrap invoke (D-21).
5. Build the invoke TLV (build_invoke_request, direction-routed).
6. Send + recv via the transport seam.
7. Parse the response (parse_invoke_response) → dict.
8. Apply DATE/TIME post-processing (D-24): str → datetime.date/time or None.
9. Always restore READY in finally (even on exception).
Raises:
AbapApplicationError: propagated from parse_invoke_response.
AbapSystemFailure: propagated from parse_invoke_response.
CommunicationError: wraps OSError and EOFError from the transport (CLIENT-06).
ValueError: propagated for malformed response TLV.
Credentials are never logged (threat T-04-CRED).
For classic TCP connections, delegates to the async core via _LoopThread (D-07).
"""
if self._async_conn is not None and self._loop_thread is not None:
return cast(
dict[str, Any], self._loop_thread.run(self._async_conn.call(func_name, **params))
)
with self._lock:
# WS lazy-LOGON: in WS_PENDING the first call sends LOGON+GFI combined
# (inside _call_bootstrap), then sends the actual function as a subsequent
# invoke. In all other states the normal READY guard applies.
ws_pending = self._is_ws() and self._session.state is SessionState.WS_PENDING
if not ws_pending:
self._session._require_state(SessionState.READY)
self._session.mark_in_call()
# Metrics were recorded only on the async core, which classic TCP
# delegates to. The wRFC and SNC paths run here instead, so a
# ConnectionMetrics on either reported zero calls however many were
# made -- a metric that is quietly absent is worse than one that is
# obviously missing, because a dashboard showing nothing looks like
# an idle connection rather than a broken counter.
_started = time.perf_counter()
self._last_server_duration_s = None
_sent_before = getattr(self._transport, "bytes_sent", 0)
_received_before = getattr(self._transport, "bytes_received", 0)
_failed = True
try:
# WS_PENDING fast-path: if the target function is in _WRFC_BUILTIN_DESCS,
# name it in the LOGON frame directly -- the LOGON runs the function
# it names, so this avoids GFI and answers in one round-trip.
if ws_pending:
builtin = _WRFC_BUILTIN_DESCS.get(func_name.upper())
if builtin is not None:
_direct = self._ws_direct_logon_call(
func_name.upper(), builtin, dict(params)
)
_failed = False
return _direct
# Fetch FunctionDesc (cache or bootstrap round-trip, D-21).
# In WS_PENDING, _call_bootstrap sends LOGON+RFC_GET_FUNCTION_INTERFACE
# and advances the session to READY before returning.
desc = get_function_desc(self, func_name, cache=self._cache)
# After get_function_desc, session is READY regardless of the ws_pending
# path. Mark IN_CALL now for the actual function invoke that follows.
if ws_pending:
self._session.mark_in_call()
if self._is_ws():
# wRFC: raw-TLV invoke over WebSocket (no GW header, no COM_HEAD).
frame = _build_ws_invoke_frame(func_name, desc, dict(params))
with _fail_closed(self._session, func_name):
self._transport.send_message(frame)
response = _join_response_frames(self._transport.recv_message, func_name)
self._last_server_duration_s = extract_server_duration(response)
result = parse_invoke_response(response, desc)
else:
# Classic GW-framed invoke (TCP / SNC).
call_params = _filter_call_params(
func_name,
desc,
dict(params),
strict=self._strict_params,
seen=self._dropped_params_seen,
)
request_tlv = build_invoke_request(func_name, desc, call_params)
dm_names = dm_table_ids(desc, call_params)
handle = self._session.handle or b" "
request = self._build_invoke_frame(handle, request_tlv)
with _fail_closed(self._session, func_name):
self._send_invoke_frame(request)
tlv_response = _join_response_frames(
self._transport.recv_message, func_name
)
self._last_server_duration_s = extract_server_duration(tlv_response)
result = parse_invoke_response(tlv_response, desc, dm_names)
result = _convert_date_time_fields(result, desc)
_failed = False
return result
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
finally:
# Recorded on the failure path too: a view that counts only
# successes hides exactly the trend worth alerting on.
self.metrics.record(
CallStats(
func_name=func_name,
duration_s=time.perf_counter() - _started,
request_bytes=getattr(self._transport, "bytes_sent", 0) - _sent_before,
response_bytes=(
getattr(self._transport, "bytes_received", 0) - _received_before
),
failed=_failed,
server_duration_s=self._last_server_duration_s,
)
)
# Only flip IN_CALL → READY; skip if state is WS_PENDING (auth failed
# before mark_in_call was ever called) or READY (post-exception cleanup).
if self._session.state is SessionState.IN_CALL:
self._session.mark_ready()
# ------------------------------------------------------------------ #
# Transactional RFC (tRFC / qRFC) client methods — TRFC-01/02/04 #
# D-06: all client methods live on Connection directly. #
# ------------------------------------------------------------------ #
def create_tid(self) -> str:
"""Generate a 24-character Transaction ID (TID) for tRFC / qRFC calls.
Uses local UUID generation (NULL-handle semantics per SDK type definitions):
this method does NOT require an open connection and may be called before
``connect()`` or after the connection is closed.
The returned TID is derived from ``uuid4().hex[:24].upper()``. UUID-hex
characters (``0-9A-F``) are a strict subset of the confirmed RFC TID
alphabet (``ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789/_=@-``;),
so the TID is always valid on the wire. The authentic SDK format uses
IP+PID+time+counter encoding, but SAP accepts any string in the alphabet
(range check only notes, Plan 06-01 SUMMARY Assumption A1).
Returns:
A 24-character uppercase string suitable for use as a TID.
Source: SDK type definitions-2224 (RfcGetTransactionID NULL-handle branch),
protocol analysis.
"""
return uuid.uuid4().hex[:24].upper()
def call_transactional(
self,
func_name: str,
*,
tid: str,
queue: str | None = None,
**params: object,
) -> None:
"""Submit a tRFC (or qRFC) call carrying the confirmed call-type marker.
Sends a synchronous RFC invoke of ``ARFC_DEST_SHIP`` with the TID encoded
as a CHAR parameter (UTF-16LE, 24 chars = 48 bytes — Pitfall 4). The
function-name TLV (0x0102) carries ``ARFC_DEST_SHIP``, which IS the
call-type discriminator on the server side (protocol analysis — no separate discriminator byte).
For qRFC (``queue`` is not None): the queue name is included as an
additional parameter in the ARFCSSTATE table param, causing the server to
read a non-zero value at the queue-indicator offset 0xe58.
Returns None — tRFC has no return values by design (CONTEXT Claude's
discretion: ``call_transactional`` returns None rather than a dict because
the ARFC_DEST_SHIP response carries no EXPORTING parameters meaningful to
the caller; exactly-once delivery is signaled by absence of exception).
This method NEVER calls ``confirm_tid`` automatically (Pitfall 3 /
D-04): confirm is a SEPARATE lifecycle step (``conn.confirm_tid(tid)``).
Calling ``confirm_tid`` before verifying the submit landed removes backend
duplicate-execution protection.
Args:
func_name: The wrapped ABAP function module name (e.g.
``"STFC_CONNECTION"``). Stored as ARFCFNAM in
ARFCSSTATE.
tid: 24-char TID from the RFC alphabet.
Use ``create_tid()`` to generate one.
queue: qRFC queue name. When not None, this call becomes a
queued RFC (TRFC-04). Must be non-empty and bounded
by the protocol maximum.
**params: Additional keyword arguments (reserved for future
ARFCSDATA payload encoding; currently unused).
Raises:
ValueError: If ``tid`` is not a valid 24-char RFC TID.
CommunicationError: Wraps ``OSError`` / ``EOFError`` from the
transport (CLIENT-06 pattern).
Security (T-06-C02): TID length and alphabet are validated inside
``build_trfc_request`` before encoding. CommunicationError does not
leak transport internals beyond ``str(exc)`` (T-06-C03).
Source: SDK type definitions–2165 (RfcCreateTransaction, RfcSubmitTransaction),
docs/protocol/trfc.md §"System FM Sequence".
"""
# Classic TCP path: delegate to async core for retry behaviour (D-07).
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(
self._async_conn.call_transactional(func_name, tid=tid, queue=queue, **params)
)
return
with self._lock:
self._session._require_state(SessionState.READY)
self._session.mark_in_call()
try:
request_tlv = build_trfc_request(tid, func_name, queue=queue)
handle = self._session.handle or b" "
request = self._build_invoke_frame(handle, request_tlv)
try:
self._send_invoke_frame(request)
response = _join_response_frames(self._transport.recv_message, func_name)
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
# tRFC has no EXPORTING params, but the reply still carries the
# return code, and reading one frame and discarding it hid both
# halves of that: a refusal read as success, and any reply longer
# than one frame left its remainder in the socket for the next
# call to misparse.
raise_for_rfc_error(_strip_gw_header(response))
finally:
self._session.mark_ready()
def confirm_tid(self, tid: str) -> None:
"""Confirm a TID as a distinct lifecycle step (TRFC-02 / D-04).
Sends a synchronous RFC invoke of ``ARFC_DEST_CONFIRM``, which causes
the SAP backend to remove the TID from ARFCRSTATE and drop duplicate-
execution protection for this TID.
WARNING: After ``confirm_tid`` returns, the backend can no longer detect
duplicate calls using this TID. Only call this method
after you have verified that the ``call_transactional`` submit landed
successfully (e.g. no ``CommunicationError`` was raised).
This method is intentionally separate from ``call_transactional`` (Pitfall
3 / D-04): bundling submit + confirm in one step breaks exactly-once
delivery in three-tier failure scenarios.
Args:
tid: The same 24-char TID passed to ``call_transactional``.
Raises:
ValueError: If ``tid`` is not a valid 24-char RFC TID.
CommunicationError: Wraps ``OSError`` / ``EOFError`` from the
transport.
Source: SDK type definitions (RfcConfirmTransactionID),
protocol analysis (ARFC_DEST_CONFIRM branch).
"""
# Classic TCP path: delegate to async core (D-07).
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(self._async_conn.confirm_tid(tid))
return
with self._lock:
self._session._require_state(SessionState.READY)
self._session.mark_in_call()
try:
request_tlv = build_trfc_confirm_request(tid)
handle = self._session.handle or b" "
request = self._build_invoke_frame(handle, request_tlv)
try:
self._send_invoke_frame(request)
response = _join_response_frames(
self._transport.recv_message, "ARFC_DEST_CONFIRM"
)
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
raise_for_rfc_error(_strip_gw_header(response))
finally:
self._session.mark_ready()
# ------------------------------------------------------------------ #
# bgRFC client methods — TRFC-05/06 #
# D-06: all client methods live on Connection directly. #
# ------------------------------------------------------------------ #
def create_unit(
self,
uid: str | None = None,
queues: list[str] | None = None,
) -> _UnitHandle:
"""Create a bgRFC unit context manager (TRFC-05 / D-05).
Returns a one-shot context manager (``_UnitHandle``) that buffers
``unit.call("FM", **params)`` invocations. On ``__exit__`` with no
exception, the buffered calls are submitted as a single atomic LUW
via BGRFC_DEST_SHIP. On exception inside the with-block, the unit
is abandoned and NO submit frame is sent (Pitfall 6).
Unit type (Pitfall 5):
- ``'T'`` when ``queues`` is empty or None (synchronous unit)
- ``'Q'`` when ``queues`` is non-empty (queued unit)
The type is stored on the handle so ``confirm_unit`` / ``get_unit_state``
can pass the correct ``RFC_UNIT_IDENTIFIER`` to the backend.
UnitID generation: when ``uid`` is None, generates a 32-char uppercase
hex UnitID via ``uuid4().hex.upper()`` (NULL-handle semantics,
SDK type definitions-2224 the UUID formatter path).
Args:
uid: 32-char uppercase hex UnitID; generated if None.
queues: List of queue names. Empty/None → unit_type 'T'.
Returns:
``_UnitHandle`` context manager. Use as::
with conn.create_unit(queues=["Q1"]) as unit:
unit.call("FM1", PARAM=val)
unit.call("FM2", PARAM=val)
# On clean exit → BGRFC_DEST_SHIP frame submitted atomically.
# On exception → unit abandoned, no submit.
Source: SDK type definitions (RfcCreateUnit), 2272 (RfcInvokeInUnit),
2303 (RfcSubmitUnit), D-05 context-manager API.
"""
if uid is None:
uid = uuid.uuid4().hex.upper()
unit_type = "Q" if (queues and len(queues) > 0) else "T"
return _UnitHandle(
connection=self,
uid=uid,
unit_type=unit_type,
queues=queues or [],
)
def _submit_unit(
self,
uid: str,
unit_type: str,
queues: list[str],
buffered_calls: list[bytes],
) -> None:
"""Internal: submit the buffered unit as a BGRFC_DEST_SHIP call.
Called by ``_UnitHandle.__exit__`` on clean exit (no exception).
Reuses the lock envelope + ``_build_invoke_frame`` — same pattern as
``call_transactional``.
Raises:
CommunicationError: Wraps OSError/EOFError from the transport.
"""
# Classic TCP path: delegate to async core for retry behaviour (D-07).
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(
self._async_conn._submit_unit(uid, unit_type, queues, buffered_calls)
)
return
with self._lock:
self._session._require_state(SessionState.READY)
self._session.mark_in_call()
try:
request_tlv = build_bgrfc_request(uid, unit_type, queues, buffered_calls)
handle = self._session.handle or b" "
request = self._build_invoke_frame(handle, request_tlv)
try:
self._send_invoke_frame(request)
response = _join_response_frames(
self._transport.recv_message, "BGRFC_DEST_SHIP"
)
except (OSError, EOFError) as exc:
raise CommunicationError(str(exc), original_exception=exc) from exc
# The submit has no EXPORTING params, but the reply still reports
# whether the backend took the unit. Reading one frame and
# discarding it, as this did, hid two failures: a refusal read as
# success, and a reply spanning more than one frame left the rest
# in the socket for the next call to parse as TLV -- which is how
# a later RFC_READ_TABLE came back as "malformed TLV: tag 0x2a45",
# the ASCII of an error string.
raise_for_rfc_error(_strip_gw_header(response))
finally:
self._session.mark_ready()
def confirm_unit(self, unit_id: str, unit_type: str = "T") -> None:
"""Confirm a bgRFC unit as a distinct lifecycle step (TRFC-06 / D-05).
Sends BGRFC_DEST_CONFIRM to the backend. After this call the backend
can clean up the unit state. The ``unit_type`` must match the type
used at submit time (Pitfall 5).
``RFC_UNIT_NOT_FOUND`` after confirm means the backend already cleaned
up — treat as success (anti-pattern: never resend on NOT_FOUND after
confirm, T-06-U04).
Args:
unit_id: 32-char uppercase hex UnitID.
unit_type: 'T' or 'Q' (must match submit-time unit_type, Pitfall 5).
Raises:
ValueError: If ``unit_id`` is not a valid 32-char hex UnitID.
CommunicationError: Wraps OSError/EOFError from the transport.
Source: SDK type definitions (RfcConfirmUnit),
protocol analysis (BGRFC_DEST_CONFIRM).
"""
# Classic TCP path: delegate to async core (D-07).
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(self._async_conn.confirm_unit(unit_id, unit_type))
return
# Driven through the ordinary call path. This module's signature is one
# the dictionary describes -- UNIT_ID as BYTE(16), UNIT_KIND as INT4 --
# so the normal encoder handles it. The bespoke builder this replaced
# sent parameters that do not exist: BGRFC_UNIT_ID as 32 hex characters
# in UTF-16LE, and BGRFC_UNIT_TYPE as the character 'T' or 'Q'.
self.call(
"BGRFC_DEST_CONFIRM",
UNIT_ID=bgrfc_unit_id_bytes(unit_id),
UNIT_KIND=bgrfc_unit_kind(unit_type),
)
def rollback_unit(self, unit_id: str, unit_type: str = "T") -> None:
"""Signal that a bgRFC unit should be rolled back (TRFC-06).
Informs the backend that the unit should be treated as rolled back
(re-send may be required). This is distinct from ``confirm_unit``
and does NOT remove the unit from the backend's state tables.
Args:
unit_id: 32-char uppercase hex UnitID.
unit_type: 'T' or 'Q' (must match submit-time unit_type, Pitfall 5).
Raises:
ValueError: If ``unit_id`` is not a valid 32-char hex UnitID.
CommunicationError: Wraps OSError/EOFError from the transport.
Source: SDK type definitions (RfcDestroyUnit / rollback path); D-05.
"""
# Classic TCP path: delegate to async core (D-07).
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(self._async_conn.rollback_unit(unit_id, unit_type))
return
# bgRFC rollback from the client side sends a state query/notification;
# the authoritative rollback happens on the server side (server-side
# on_rollback callback). Client-side rollback records intent and does NOT
# submit (consistent with Pitfall 3 — never bundle submit+rollback).
# This call is a no-op over the wire when the transport is not live
# (OG-06-02 gate); the pattern is documented here for completeness.
# There is no client-side rollback module. A state query is issued so the
# unit id is validated against the backend and the caller learns where the
# unit actually stands, which is the only honest thing available here.
self.get_unit_state(unit_id, unit_type)
def get_unit_state(self, unit_id: str, unit_type: str = "T") -> UnitState:
"""Query the current state of a bgRFC unit on the backend (TRFC-06).
Sends BGRFC_CHECK_UNIT_STATE_SERVER and maps the response to a
``UnitState`` enum value (SDK type definitions-332).
``RFC_UNIT_NOT_FOUND`` after a confirmed unit is treated as success
(state is already ``CONFIRMED`` — do not resend, T-06-U04).
Args:
unit_id: 32-char uppercase hex UnitID.
unit_type: 'T' or 'Q' (must match submit-time unit_type, Pitfall 5).
Returns:
A ``UnitState`` enum value.
Raises:
ValueError: If ``unit_id`` is not a valid 32-char hex UnitID.
CommunicationError: Wraps OSError/EOFError from the transport.
Source: SDK type definitions (RfcGetUnitState),
protocol analysis (BGRFC_CHECK_UNIT_STATE_SERVER).
"""
# Classic TCP path: delegate to async core (D-07).
if self._async_conn is not None and self._loop_thread is not None:
return cast(
UnitState,
self._loop_thread.run(self._async_conn.get_unit_state(unit_id, unit_type)),
)
# Driven through the ordinary call path. This module's signature is one
# the dictionary describes -- UNIT_ID as BYTE(16), UNIT_KIND as INT4 --
# so the normal encoder handles it. The bespoke builder this replaced
# sent parameters that do not exist: BGRFC_UNIT_ID as 32 hex characters
# in UTF-16LE, and BGRFC_UNIT_TYPE as the character 'T' or 'Q'.
result = self.call(
"BGRFC_CHECK_UNIT_STATE_SERVER",
UNIT_ID=bgrfc_unit_id_bytes(unit_id),
UNIT_KIND=bgrfc_unit_kind(unit_type),
)
raw = result.get("STATE")
if not isinstance(raw, int):
raise TransactionalError(
f"BGRFC_CHECK_UNIT_STATE_SERVER returned no integer STATE for {unit_id}; "
f"got {type(raw).__name__}"
)
name, recognised = unit_state_from_wire(raw)
if not recognised:
raise TransactionalError(
f"BGRFC_CHECK_UNIT_STATE_SERVER answered STATE={raw} for {unit_id}, a "
"value this library has no meaning for. Reported rather than guessed: "
"the parser this replaced answered NOT_FOUND for anything it could not "
"read, so an unrecognised state was indistinguishable from a unit the "
"backend has no record of."
)
return UnitState[name]
@staticmethod
def _parse_unit_state_response(response: bytes) -> UnitState:
"""Parse a BGRFC_CHECK_UNIT_STATE_SERVER response into a UnitState enum.
The backend returns the state as a CHAR parameter (BGRFC_STATE) in the
response TLV. Map the string value to UnitState (SDK type definitions-332).
When no recognisable state is found (offline or unknown value), return
UnitState.NOT_FOUND (safe default — caller can treat as not yet committed).
Mapping (RFC_UNIT_STATE → UnitState):
0 / 'NOT_FOUND' → UnitState.NOT_FOUND
1 / 'IN_PROCESS' → UnitState.IN_PROCESS
2 / 'COMMITTED' → UnitState.COMMITTED
3 / 'ROLLED_BACK'→ UnitState.ROLLED_BACK
4 / 'CONFIRMED' → UnitState.CONFIRMED
"""
if not response:
return UnitState.NOT_FOUND
from saprfclib.invoke import _decode_utf16le
try:
for name, val in _extract_name_value_pairs(response):
if name.upper() in ("BGRFC_STATE", "STATE", "UNIT_STATE"):
state_str = _decode_utf16le(val).strip()
_state_map = {
"0": UnitState.NOT_FOUND,
"NOT_FOUND": UnitState.NOT_FOUND,
"1": UnitState.IN_PROCESS,
"IN_PROCESS": UnitState.IN_PROCESS,
"2": UnitState.COMMITTED,
"COMMITTED": UnitState.COMMITTED,
"3": UnitState.ROLLED_BACK,
"ROLLED_BACK": UnitState.ROLLED_BACK,
"4": UnitState.CONFIRMED,
"CONFIRMED": UnitState.CONFIRMED,
}
return _state_map.get(state_str.upper(), UnitState.NOT_FOUND)
except Exception:
pass
return UnitState.NOT_FOUND
def retry_parked(self, tid: str) -> None:
"""Re-send a parked tRFC call from the durable store (D-03b — sync delegation).
Delegates to :meth:`AsyncConnection.retry_parked` via the background event
loop (D-07). Only available for classic TCP connections (``_async_conn`` is
set). Raises :class:`~saprfclib.exceptions.TransactionalError` for SNC/wRFC
paths where no async core is present.
"""
from saprfclib.exceptions import TransactionalError
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(self._async_conn.retry_parked(tid))
return
raise TransactionalError("retry_parked is only available on classic TCP connections")
def retry_parked_unit(self, unit_id: str, unit_type: str = "T") -> None:
"""Re-send a parked bgRFC unit from the durable store (D-03b — sync delegation).
Delegates to :meth:`AsyncConnection.retry_parked_unit` via the background
event loop (D-07). Only available for classic TCP connections.
"""
from saprfclib.exceptions import TransactionalError
if self._async_conn is not None and self._loop_thread is not None:
self._loop_thread.run(self._async_conn.retry_parked_unit(unit_id, unit_type))
return
raise TransactionalError("retry_parked_unit is only available on classic TCP connections")
def close(self) -> None:
"""Close the connection; safe to call in ANY state including partial/error.
Suppresses every exception from the (future) RFC-layer close frame, then
unconditionally marks the session CLOSED and closes the transport
(TRANS-06). Idempotent: closing an already-closed connection never raises.
For classic TCP connections, delegates to the async core then stops the
background event loop (D-07).
"""
if self._async_conn is not None and self._loop_thread is not None:
try:
self._loop_thread.run(self._async_conn.close())
except Exception:
pass
finally:
self._loop_thread.close()
self._async_conn = None
self._loop_thread = None
# Mark session CLOSED so subsequent ping/call raise ValueError.
self._session._state = SessionState.CLOSED
return
try:
# Phase 4 will send an RFC-layer close frame here when in a clean state.
pass
except Exception:
pass
finally:
self._session._state = SessionState.CLOSED
self._transport.close()