Attempt to using Nakama as replacement of Low-Level ENet

This commit is contained in:
2025-12-02 00:58:44 +08:00
parent b27b612989
commit ead155afed
74 changed files with 14205 additions and 23315 deletions
@@ -0,0 +1,34 @@
extends RefCounted
class_name NakamaAsyncResult
var exception : NakamaException:
set(v):
pass
get:
return get_exception()
var _ex = null
func _init(p_ex = null):
_ex = p_ex
func is_exception():
return get_exception() != null
func was_cancelled():
return is_exception() and get_exception().cancelled
func get_exception() -> NakamaException:
return _ex as NakamaException
func _to_string():
if is_exception():
return get_exception()._to_string()
return "NakamaAsyncResult<>"
static func _safe_ret(p_obj, p_type : GDScript):
if is_instance_of(p_obj, p_type):
return p_obj # Correct type
elif p_obj is NakamaException:
return p_type.new(p_obj) # It's an exception. Incapsulate it
return p_type.new(NakamaException.new()) # It's something else. generate an exception
@@ -0,0 +1 @@
uid://ckpfmnm3crkgo
@@ -0,0 +1,42 @@
extends RefCounted
# An exception generated during a request.
# Usually contains at least an error message.
class_name NakamaException
var _status_code : int = -1
var status_code : int:
set(v):
pass
get:
return _status_code
var _grpc_status_code : int = -1
var grpc_status_code : int:
set(v):
pass
get:
return _grpc_status_code
var _message : String = ""
var message : String:
set(v):
pass
get:
return _message
var _cancelled : bool = false
var cancelled : bool:
set(v):
pass
get:
return _cancelled
func _init(p_message : String = "", p_status_code : int = -1, p_grpc_status_code : int = -1, p_cancelled : bool = false):
_status_code = p_status_code
_grpc_status_code = p_grpc_status_code
_message = p_message
_cancelled = p_cancelled
func _to_string() -> String:
return "NakamaException(StatusCode={%s}, Message='{%s}', GrpcStatusCode={%s})" % [_status_code, _message, _grpc_status_code]
@@ -0,0 +1 @@
uid://dl5bqgn61sifg
@@ -0,0 +1,38 @@
extends RefCounted
class_name NakamaLogger
enum LOG_LEVEL {NONE, ERROR, WARNING, INFO, VERBOSE, DEBUG}
var _level = LOG_LEVEL.ERROR
var _module = "Nakama"
func _init(p_module : String = "Nakama", p_level : int = LOG_LEVEL.ERROR):
_level = p_level
_module = p_module
func _log(level : int, msg):
if level <= _level:
if level == LOG_LEVEL.ERROR:
printerr("=== %s : ERROR === %s" % [_module, str(msg)])
else:
var what = "=== UNKNOWN === "
for k in LOG_LEVEL:
if level == LOG_LEVEL[k]:
what = "=== %s : %s === " % [_module, k]
break
print(what + str(msg))
func error(msg):
_log(LOG_LEVEL.ERROR, msg)
func warning(msg):
_log(LOG_LEVEL.WARNING, msg)
func info(msg):
_log(LOG_LEVEL.INFO, msg)
func verbose(msg):
_log(LOG_LEVEL.VERBOSE, msg)
func debug(msg):
_log(LOG_LEVEL.DEBUG, msg)
@@ -0,0 +1 @@
uid://bpe0ljoah7ssn
@@ -0,0 +1,363 @@
extends RefCounted
class_name NakamaMultiplayerBridge
enum MatchState {
DISCONNECTED,
JOINING,
CONNECTED,
SOCKET_CLOSED,
}
enum MetaMessageType {
CLAIM_HOST,
ASSIGN_PEER_ID,
}
# Read-only variables.
var _nakama_socket: NakamaSocket
var nakama_socket: NakamaSocket:
get: return _nakama_socket
set(_v): pass
var _match_state: int = MatchState.DISCONNECTED
var match_state: int:
get: return _match_state
set(_v): pass
var _match_id := ''
var match_id: String:
get: return _match_id
set(_v): pass
var _multiplayer_peer: NakamaMultiplayerPeer = NakamaMultiplayerPeer.new()
var multiplayer_peer: NakamaMultiplayerPeer:
get: return _multiplayer_peer
set(_v): pass
# Configuration that can be set by the developer.
var meta_op_code: int = 9001
var rpc_op_code: int = 9002
# Internal variables.
var _my_session_id: String
var _my_peer_id: int = 0
var _id_map := {}
var _users := {}
var _matchmaker_ticket := ''
class User extends RefCounted:
var presence
var peer_id: int = 0
func _init(p_presence) -> void:
presence = p_presence
signal match_join_error (exception)
signal match_joined ()
func _set_readonly(_value) -> void:
pass
func _init(p_nakama_socket: NakamaSocket) -> void:
_nakama_socket = p_nakama_socket
_nakama_socket.received_match_presence.connect(self._on_nakama_socket_received_match_presence)
_nakama_socket.received_matchmaker_matched.connect(self._on_nakama_socket_received_matchmaker_matched)
_nakama_socket.received_match_state.connect(self._on_nakama_socket_received_match_state)
_nakama_socket.closed.connect(self._on_nakama_socket_closed)
_multiplayer_peer.packet_generated.connect(self._on_multiplayer_peer_packet_generated)
_multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTING)
func create_match() -> void:
if _match_state != MatchState.DISCONNECTED:
push_error("Cannot create match when state is %s" % MatchState.keys()[_match_state])
return
_match_state = MatchState.JOINING
multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTING)
var res = await _nakama_socket.create_match_async()
if res.is_exception():
match_join_error.emit(res.get_exception())
leave()
return
_setup_match(res)
_setup_host()
func join_match(p_match_id: String) -> void:
if _match_state != MatchState.DISCONNECTED:
push_error("Cannot join match when state is %s" % MatchState.keys()[_match_state])
return
_match_state = MatchState.JOINING
multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTING)
var res = await _nakama_socket.join_match_async(p_match_id)
if res.is_exception():
match_join_error.emit(res.get_exception())
leave()
return
_setup_match(res)
func join_named_match(_match_name: String) -> void:
if _match_state != MatchState.DISCONNECTED:
push_error("Cannot join match when state is %s" % MatchState.keys()[_match_state])
return
_match_state = MatchState.JOINING
multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTING)
var res = await _nakama_socket.create_match_async(_match_name)
if res.is_exception():
match_join_error.emit(res.get_exception())
leave()
return
_setup_match(res)
if res.size == 0 or (res.size == 1 and res.presences.size() == 0):
_setup_host()
func start_matchmaking(ticket) -> void:
if _match_state != MatchState.DISCONNECTED:
push_error("Cannot start matchmaking when state is %s" % MatchState.keys()[_match_state])
return
if ticket.is_exception():
push_error("Ticket with exception passed into start_matchmaking()")
return
_match_state = MatchState.JOINING
multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTING)
_matchmaker_ticket = ticket.ticket
func _on_nakama_socket_received_matchmaker_matched(matchmaker_matched) -> void:
if _matchmaker_ticket != matchmaker_matched.ticket:
return
# Get a list of sorted session ids.
var session_ids := []
for matchmaker_user in matchmaker_matched.users:
session_ids.append(matchmaker_user.presence.session_id)
session_ids.sort()
var res = await _nakama_socket.join_matched_async(matchmaker_matched)
if res.is_exception():
match_join_error.emit(res.get_exception())
leave()
return
_setup_match(res)
# If our session is the first alphabetically, then we'll be the host.
if _my_session_id == session_ids[0]:
_setup_host()
# Add all of the existing peers.
for presence in res.presences:
if presence.session_id != _my_session_id:
_host_add_peer(presence)
func _on_nakama_socket_closed() -> void:
match_state = MatchState.SOCKET_CLOSED
_cleanup()
func get_user_presence_for_peer(peer_id: int) -> NakamaRTAPI.UserPresence:
var session_id = _id_map.get(peer_id)
if session_id == null:
return null
var user = _users.get(session_id)
if user == null:
return null
return user.presence
func leave() -> void:
if _match_state == MatchState.DISCONNECTED:
return
_match_state = MatchState.DISCONNECTED
if _match_id:
await _nakama_socket.leave_match_async(_match_id)
if _matchmaker_ticket:
await _nakama_socket.remove_matchmaker_async(_matchmaker_ticket)
_cleanup()
func _cleanup() -> void:
for peer_id in _id_map:
multiplayer_peer.peer_disconnected.emit(peer_id)
_match_id = ''
_matchmaker_ticket = ''
_my_session_id = ''
_my_peer_id = 0
_id_map.clear()
_users.clear()
_multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_DISCONNECTED)
func _setup_match(res) -> void:
_match_id = res.match_id
_my_session_id = res.self_user.session_id
_users[_my_session_id] = User.new(res.self_user)
for presence in res.presences:
if not _users.has(presence.session_id):
_users[presence.session_id] = User.new(presence)
func _setup_host() -> void:
# Claim id 1 and start the match.
_my_peer_id = 1
_map_id_to_session(1, _my_session_id)
_match_state = MatchState.CONNECTED
_multiplayer_peer.initialize(_my_peer_id)
match_joined.emit()
func _generate_id(session_id: String) -> int:
# Peer ids can only be positive 32-bit signed integers.
var peer_id: int = session_id.hash() & 0x7FFFFFFF
# If this peer id is already taken, try to find another.
while peer_id <= 1 or _id_map.has(peer_id):
peer_id += 1
if peer_id > 0x7FFFFFFF or peer_id <= 0:
peer_id = randi() & 0x7FFFFFFF
return peer_id
func _map_id_to_session(peer_id: int, session_id: String) -> void:
_id_map[peer_id] = session_id
_users[session_id].peer_id = peer_id
func _host_add_peer(presence) -> void:
var peer_id = _generate_id(presence.session_id)
_map_id_to_session(peer_id, presence.session_id)
# Tell them we are the host.
_nakama_socket.send_match_state_async(_match_id, meta_op_code, JSON.stringify({
type = MetaMessageType.CLAIM_HOST,
}), [presence])
# Tell them about all the other connected peers.
for other_peer_id in _id_map:
var other_session_id = _id_map[other_peer_id]
if other_session_id == presence.session_id or other_session_id == _my_session_id:
continue
_nakama_socket.send_match_state_async(_match_id, meta_op_code, JSON.stringify({
type = MetaMessageType.ASSIGN_PEER_ID,
session_id = other_session_id,
peer_id = other_peer_id,
}), [presence])
# Assign them a peer_id (tell everyone about it).
_nakama_socket.send_match_state_async(_match_id, meta_op_code, JSON.stringify({
type = MetaMessageType.ASSIGN_PEER_ID,
session_id = presence.session_id,
peer_id = peer_id,
}))
_multiplayer_peer.peer_connected.emit(peer_id)
func _on_nakama_socket_received_match_presence(event) -> void:
if _match_state == MatchState.DISCONNECTED:
return
if event.match_id != _match_id:
return
for presence in event.joins:
if not _users.has(presence.session_id):
_users[presence.session_id] = User.new(presence)
# If we are the host, and they don't yet have a peer id, then let's
# generate a new id for them and send all the necessary messages.
if _my_peer_id == 1 and _users[presence.session_id].peer_id == 0:
_host_add_peer(presence)
for presence in event.leaves:
if not _users.has(presence.session_id):
continue
var peer_id = _users[presence.session_id].peer_id
_multiplayer_peer.peer_disconnected.emit(peer_id)
_users.erase(presence.session_id)
_id_map.erase(peer_id)
func _parse_json(data: String):
var json = JSON.new()
if json.parse(data) != OK:
return null
var content = json.get_data()
if not content is Dictionary:
return null
return content
func _on_nakama_socket_received_match_state(data) -> void:
if _match_state == MatchState.DISCONNECTED:
return
if data.match_id != _match_id:
return
if data.op_code == meta_op_code:
var content = _parse_json(data.data)
if content == null:
return
var type = content['type']
#print ("RECEIVED: ", content)
if type == MetaMessageType.CLAIM_HOST:
if _id_map.has(1):
# @todo Can we mediate this dispute?
push_error("User %s claiming to be host, when user %s has already claimed it" % [data.presence.session_id, _id_map[1]])
else:
_map_id_to_session(1, data.presence.session_id)
return
# Ensure that any meta messages are coming from the host!
if data.presence.session_id != _id_map[1]:
push_error("Received meta message from user %s who isn't the host: %s" % [data.presence.session_id, content])
return
if type == MetaMessageType.ASSIGN_PEER_ID:
var session_id = content['session_id']
var peer_id = content['peer_id']
if _users.has(session_id) and _users[session_id].peer_id != 0:
push_error("Attempting to assign peer id %s to %s which already has id %s" % [
peer_id,
session_id,
_users[session_id].peer_id,
])
return
_map_id_to_session(peer_id, session_id)
if _my_session_id == session_id:
_match_state = MatchState.CONNECTED
_multiplayer_peer.initialize(peer_id)
_multiplayer_peer.set_connection_status(MultiplayerPeer.CONNECTION_CONNECTED)
match_joined.emit()
_multiplayer_peer.peer_connected.emit(1)
else:
_multiplayer_peer.peer_connected.emit(peer_id)
else:
_nakama_socket.logger.error("Received meta message with unknown type: %s" % type)
elif data.op_code == rpc_op_code:
var from_session_id: String = data.presence.session_id
if not _users.has(from_session_id) or _users[from_session_id].peer_id == 0:
push_error("Received RPC from %s which isn't assigned a peer id" % data.presence.session_id)
return
var from_peer_id = _users[from_session_id].peer_id
_multiplayer_peer.deliver_packet(data.binary_data, from_peer_id)
func _on_multiplayer_peer_packet_generated(peer_id: int, buffer: PackedByteArray) -> void:
if match_state == MatchState.CONNECTED:
var target_presences = null
if peer_id > 0:
if not _id_map.has(peer_id):
push_error("Attempting to send RPC to unknown peer id: %s" % peer_id)
return
target_presences = [ _users[_id_map[peer_id]].presence ]
_nakama_socket.send_match_state_raw_async(_match_id, rpc_op_code, buffer, target_presences)
else:
push_error("RPC sent while the NakamaMultiplayerBridge isn't connected!")
@@ -0,0 +1 @@
uid://ntxj8ansnmt
@@ -0,0 +1,97 @@
extends MultiplayerPeerExtension
class_name NakamaMultiplayerPeer
const MAX_PACKET_SIZE := 1 << 24
var _self_id := 0
var _connection_status: ConnectionStatus = CONNECTION_DISCONNECTED
var _refusing_new_connections := false
var _target_id := 0
class Packet extends RefCounted:
var data: PackedByteArray
var from: int
func _init(p_data: PackedByteArray, p_from: int) -> void:
data = p_data
from = p_from
var _incoming_packets := []
signal packet_generated (peer_id, buffer)
func _get_packet_script() -> PackedByteArray:
if _incoming_packets.size() == 0:
return PackedByteArray()
return _incoming_packets.pop_front().data
func _get_packet_mode() -> TransferMode:
return TRANSFER_MODE_RELIABLE
func _get_packet_channel() -> int:
return 0
func _put_packet_script(p_buffer: PackedByteArray) -> Error:
packet_generated.emit(_target_id, p_buffer)
return OK
func _get_available_packet_count() -> int:
return _incoming_packets.size()
func _get_max_packet_size() -> int:
return MAX_PACKET_SIZE
func _set_transfer_channel(p_channel) -> void:
pass
func _get_transfer_channel() -> int:
return 0
func _set_transfer_mode(p_mode: TransferMode) -> void:
pass
func _get_transfer_mode() -> TransferMode:
return TRANSFER_MODE_RELIABLE
func _set_target_peer(p_peer_id: int) -> void:
_target_id = p_peer_id
func _get_packet_peer() -> int:
if _connection_status != CONNECTION_CONNECTED:
return 1
if _incoming_packets.size() == 0:
return 1
return _incoming_packets[0].from
func _is_server() -> bool:
return _self_id == 1
func _poll() -> void:
pass
func _get_unique_id() -> int:
return _self_id
func _set_refuse_new_connections(p_enable: bool) -> void:
_refusing_new_connections = p_enable
func _is_refusing_new_connections() -> bool:
return _refusing_new_connections
func _get_connection_status() -> ConnectionStatus:
return _connection_status
func initialize(p_self_id: int) -> void:
if _connection_status != CONNECTION_CONNECTING:
return
_self_id = p_self_id
if _self_id == 1:
_connection_status = CONNECTION_CONNECTED
func set_connection_status(p_connection_status: int) -> void:
_connection_status = p_connection_status
func deliver_packet(p_data: PackedByteArray, p_from_peer_id: int) -> void:
var packet = Packet.new(p_data, p_from_peer_id);
_incoming_packets.push_back(packet)
@@ -0,0 +1 @@
uid://bqi4fibji5beq
@@ -0,0 +1,160 @@
extends RefCounted
class_name NakamaSerializer
static func serialize(p_obj : Object) -> Dictionary:
var out = {}
var schema = p_obj.get("_SCHEMA")
if schema == null:
return {} # No schema defined
for k in schema:
var prop = schema[k]
var val = p_obj.get(prop["name"])
if val == null:
continue
var type = prop["type"]
var content = prop.get("content", TYPE_NIL)
if typeof(content) == TYPE_STRING:
content = TYPE_OBJECT
var val_type = typeof(val)
match val_type:
TYPE_OBJECT: # Simple objects
out[k] = serialize(val)
TYPE_ARRAY: # Array of objects
var arr = []
for e in val:
if typeof(e) != TYPE_OBJECT:
continue
arr.append(serialize(e))
out[k] = arr
TYPE_PACKED_INT32_ARRAY, TYPE_PACKED_STRING_ARRAY: # Array of ints, bools, or strings
var arr = []
for e in val:
if content == TYPE_BOOL:
e = bool(e)
if typeof(e) != content:
continue
arr.append(e)
out[k] = arr
TYPE_DICTIONARY: # Maps
var dict = {}
if content == TYPE_OBJECT: # Map of objects
for l in val:
if typeof(val[l]) != TYPE_OBJECT:
continue
dict[l] = serialize(val[l])
else: # Map of simple types
for l in val:
var e = val[l]
if content == TYPE_FLOAT:
e = float(e)
elif content == TYPE_INT:
e = int(e)
elif content == TYPE_BOOL:
e = bool(e)
if typeof(e) != content:
continue
dict[l] = e
out[k] = dict
_:
out[k] = val
return out
static func deserialize(p_ns : GDScript, p_cls_name : String, p_dict : Dictionary) -> Object:
var cls : GDScript = p_ns.get(p_cls_name)
var schema = cls.get("_SCHEMA")
if schema == null:
return NakamaException.new() # No schema defined
var obj = cls.new()
for k in schema:
var prop = schema[k]
var pname = prop["name"]
var type = prop["type"]
var required = prop["required"]
var content = prop.get("content", TYPE_NIL)
var type_cmp = type
if typeof(type) == TYPE_STRING: # A class
type_cmp = TYPE_DICTIONARY
if type_cmp == TYPE_PACKED_STRING_ARRAY or type_cmp == TYPE_PACKED_INT32_ARRAY: # A specialized array
type_cmp = TYPE_ARRAY
var content_cmp = content
if typeof(content) == TYPE_STRING: # A dictionary or array of classes
content_cmp = TYPE_DICTIONARY
var val = p_dict.get(k, null)
# Ints might and up being recognized as floats. Change that if needed
if type_cmp == TYPE_INT:
if typeof(val) == TYPE_FLOAT:
val = int(val)
elif typeof(val) == TYPE_STRING and val.is_valid_int():
val = val.to_int()
if typeof(val) == type_cmp:
if typeof(type) == TYPE_STRING:
obj.set(pname, deserialize(p_ns, type, val))
elif type_cmp == TYPE_DICTIONARY:
var v = {}
for l in val:
if typeof(content) == TYPE_STRING:
v[l] = deserialize(p_ns, content, val[l])
elif content == TYPE_FLOAT:
v[l] = float(val[l])
elif content == TYPE_INT:
v[l] = int(val[l])
elif content == TYPE_BOOL:
v[l] = bool(val[l])
else:
v[l] = str(val[l])
obj.set(pname, v)
elif type_cmp == TYPE_ARRAY:
var v
match content:
TYPE_INT, TYPE_BOOL: v = PackedInt32Array()
TYPE_STRING: v = PackedStringArray()
_: v = Array()
for e in val:
if typeof(content) == TYPE_STRING:
v.append(deserialize(p_ns, content, e))
elif content == TYPE_FLOAT:
v.append(float(e))
elif content == TYPE_INT:
v.append(int(e))
elif content == TYPE_BOOL:
v.append(bool(e))
else:
v.append(str(e))
obj.set(pname, v)
else:
obj.set(pname, val)
elif required:
obj._ex = NakamaException.new("ERROR [%s]: Missing or invalid required prop %s = %s:\n\t%s" % [p_cls_name, prop, p_dict.get(k), p_dict])
return obj
return obj
###
# Compatibility with Godot 3.1 which does not expose String.http_escape
###
const HEX = ["0","1","2","3","4","5","6","7","8","9","A","B","C","D","E","F"]
static func escape_http(p_str : String) -> String:
var out : String = ""
for o in p_str:
if (o == '.' or o == '-' or o == '_' or o == '~' or
(o >= 'a' and o <= 'z') or
(o >= 'A' and o <= 'Z') or
(o >= '0' and o <= '9')):
out += o
else:
for b in o.to_utf8_buffer():
out += "%%%s" % to_hex(b)
return out
static func to_hex(p_val : int) -> String:
var v := p_val
var o := ""
while v != 0:
o = HEX[v % 16] + o
v /= 16
return o
@@ -0,0 +1 @@
uid://bllhuxmb0a5ml