1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
# A handful of classes and functions to support the generated data structures.
# This would be a good candidate for isolating in its own ffi-support lib.
class InternalError(Exception):
pass
class _UniffiRustCallStatus(ctypes.Structure):
"""
Error runtime.
"""
_fields_ = [
("code", ctypes.c_int8),
("error_buf", _UniffiRustBuffer),
]
# These match the values from the uniffi::rustcalls module
CALL_SUCCESS = 0
CALL_ERROR = 1
CALL_PANIC = 2
def __str__(self):
if self.code == _UniffiRustCallStatus.CALL_SUCCESS:
return "_UniffiRustCallStatus(CALL_SUCCESS)"
elif self.code == _UniffiRustCallStatus.CALL_ERROR:
return "_UniffiRustCallStatus(CALL_ERROR)"
elif self.code == _UniffiRustCallStatus.CALL_PANIC:
return "_UniffiRustCallStatus(CALL_PANIC)"
else:
return "_UniffiRustCallStatus(<invalid code>)"
def _rust_call(fn, *args):
# Call a rust function
return _rust_call_with_error(None, fn, *args)
def _rust_call_with_error(error_ffi_converter, fn, *args):
# Call a rust function and handle any errors
#
# This function is used for rust calls that return Result<> and therefore can set the CALL_ERROR status code.
# error_ffi_converter must be set to the _UniffiConverter for the error class that corresponds to the result.
call_status = _UniffiRustCallStatus(code=_UniffiRustCallStatus.CALL_SUCCESS, error_buf=_UniffiRustBuffer(0, 0, None))
args_with_error = args + (ctypes.byref(call_status),)
result = fn(*args_with_error)
_uniffi_check_call_status(error_ffi_converter, call_status)
return result
def _uniffi_check_call_status(error_ffi_converter, call_status):
if call_status.code == _UniffiRustCallStatus.CALL_SUCCESS:
pass
elif call_status.code == _UniffiRustCallStatus.CALL_ERROR:
if error_ffi_converter is None:
call_status.error_buf.free()
raise InternalError("_rust_call_with_error: CALL_ERROR, but error_ffi_converter is None")
else:
raise error_ffi_converter.lift(call_status.error_buf)
elif call_status.code == _UniffiRustCallStatus.CALL_PANIC:
# When the rust code sees a panic, it tries to construct a _UniffiRustBuffer
# with the message. But if that code panics, then it just sends back
# an empty buffer.
if call_status.error_buf.len > 0:
msg = _UniffiConverterString.lift(call_status.error_buf)
else:
msg = "Unknown rust panic"
raise InternalError(msg)
else:
raise InternalError("Invalid _UniffiRustCallStatus code: {}".format(
call_status.code))
# A function pointer for a callback as defined by UniFFI.
# Rust definition `fn(handle: u64, method: u32, args: _UniffiRustBuffer, buf_ptr: *mut _UniffiRustBuffer) -> int`
_UNIFFI_FOREIGN_CALLBACK_T = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_ulonglong, ctypes.c_ulong, ctypes.POINTER(ctypes.c_char), ctypes.c_int, ctypes.POINTER(_UniffiRustBuffer))
# UniFFI future continuation
_UNIFFI_FUTURE_CONTINUATION_T = ctypes.CFUNCTYPE(None, ctypes.c_size_t, ctypes.c_int8)
|
