439 lines
13 KiB
Python
439 lines
13 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import socket
|
||
|
|
import threading
|
||
|
|
import traceback
|
||
|
|
|
||
|
|
import mod.server.extraServerApi as serverApi
|
||
|
|
|
||
|
|
from .Config import GET_MCP_BRIDGE_STATE_PATH
|
||
|
|
|
||
|
|
try:
|
||
|
|
basestring
|
||
|
|
except NameError:
|
||
|
|
basestring = str
|
||
|
|
|
||
|
|
try:
|
||
|
|
integer_types = (int, long)
|
||
|
|
except NameError:
|
||
|
|
integer_types = (int,)
|
||
|
|
|
||
|
|
_BRIDGE = None
|
||
|
|
_SERVER_ACTIVE = False
|
||
|
|
_SERVER_TASKS = []
|
||
|
|
_SERVER_TASKS_LOCK = threading.Lock()
|
||
|
|
|
||
|
|
MAX_PENDING_TASKS = 32
|
||
|
|
MAX_REQUEST_BYTES = 1024 * 1024
|
||
|
|
|
||
|
|
|
||
|
|
class BridgeError(Exception):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def _is_int(value):
|
||
|
|
return isinstance(value, integer_types) and not isinstance(value, bool)
|
||
|
|
|
||
|
|
|
||
|
|
def _event_is_set(event):
|
||
|
|
if hasattr(event, "is_set"):
|
||
|
|
return event.is_set()
|
||
|
|
return event.isSet()
|
||
|
|
|
||
|
|
|
||
|
|
def _create_daemon_thread(target, args=()):
|
||
|
|
thread = threading.Thread(target=target, args=args)
|
||
|
|
try:
|
||
|
|
thread.daemon = True
|
||
|
|
except Exception:
|
||
|
|
thread.setDaemon(True)
|
||
|
|
return thread
|
||
|
|
|
||
|
|
|
||
|
|
def _to_text(data):
|
||
|
|
if isinstance(data, bytes):
|
||
|
|
return data.decode("utf-8")
|
||
|
|
return data
|
||
|
|
|
||
|
|
|
||
|
|
def _send_json_line(sock, payload):
|
||
|
|
data = json.dumps(payload, separators=(",", ":")) + "\n"
|
||
|
|
if not isinstance(data, bytes):
|
||
|
|
data = data.encode("utf-8")
|
||
|
|
sock.sendall(data)
|
||
|
|
|
||
|
|
|
||
|
|
def _require_params(params):
|
||
|
|
if not isinstance(params, dict):
|
||
|
|
raise BridgeError("place_block params must be an object")
|
||
|
|
|
||
|
|
for key in ("x", "y", "z"):
|
||
|
|
if key not in params or not _is_int(params.get(key)):
|
||
|
|
raise BridgeError("place_block.%s must be an integer" % key)
|
||
|
|
|
||
|
|
name = params.get("name")
|
||
|
|
if not isinstance(name, basestring) or not name.strip() or ":" not in name:
|
||
|
|
raise BridgeError("place_block.name must be a non-empty namespaced block id")
|
||
|
|
|
||
|
|
dimension = params.get("dimension", 0)
|
||
|
|
if not _is_int(dimension):
|
||
|
|
raise BridgeError("place_block.dimension must be an integer")
|
||
|
|
|
||
|
|
old_block_handling = params.get("old_block_handling", 0)
|
||
|
|
if not _is_int(old_block_handling) or old_block_handling not in (0, 1, 2):
|
||
|
|
raise BridgeError("place_block.old_block_handling must be 0, 1, or 2")
|
||
|
|
|
||
|
|
aux = params.get("aux", None)
|
||
|
|
states = params.get("states", None)
|
||
|
|
if aux is not None and states is not None:
|
||
|
|
raise BridgeError("place_block.aux and place_block.states cannot both be provided")
|
||
|
|
if aux is not None and not _is_int(aux):
|
||
|
|
raise BridgeError("place_block.aux must be an integer")
|
||
|
|
if states is not None and not isinstance(states, dict):
|
||
|
|
raise BridgeError("place_block.states must be an object")
|
||
|
|
|
||
|
|
is_legacy = params.get("is_legacy", True)
|
||
|
|
if not isinstance(is_legacy, bool):
|
||
|
|
raise BridgeError("place_block.is_legacy must be a boolean")
|
||
|
|
|
||
|
|
update_neighbors = params.get("update_neighbors", True)
|
||
|
|
if not isinstance(update_neighbors, bool):
|
||
|
|
raise BridgeError("place_block.update_neighbors must be a boolean")
|
||
|
|
|
||
|
|
return {
|
||
|
|
"x": params["x"],
|
||
|
|
"y": params["y"],
|
||
|
|
"z": params["z"],
|
||
|
|
"name": name,
|
||
|
|
"dimension": dimension,
|
||
|
|
"aux": aux,
|
||
|
|
"states": states,
|
||
|
|
"old_block_handling": old_block_handling,
|
||
|
|
"is_legacy": is_legacy,
|
||
|
|
"update_neighbors": update_neighbors,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _place_block_on_server(request):
|
||
|
|
name = request["name"]
|
||
|
|
aux = request["aux"]
|
||
|
|
states = request["states"]
|
||
|
|
if states is not None:
|
||
|
|
block_state = serverApi.GetEngineCompFactory().CreateBlockState(serverApi.GetLevelId())
|
||
|
|
aux = block_state.GetBlockAuxValueFromStates(name, states)
|
||
|
|
if aux == -1:
|
||
|
|
raise BridgeError("invalid block states for %s" % name)
|
||
|
|
elif aux is None:
|
||
|
|
aux = 0
|
||
|
|
|
||
|
|
x = request["x"]
|
||
|
|
y = request["y"]
|
||
|
|
z = request["z"]
|
||
|
|
dimension = request["dimension"]
|
||
|
|
old_block_handling = request["old_block_handling"]
|
||
|
|
is_legacy = request["is_legacy"]
|
||
|
|
update_neighbors = request["update_neighbors"]
|
||
|
|
|
||
|
|
block_info = serverApi.GetEngineCompFactory().CreateBlockInfo(serverApi.GetLevelId())
|
||
|
|
changed = block_info.SetBlockNew(
|
||
|
|
(x, y, z),
|
||
|
|
{"name": name, "aux": aux},
|
||
|
|
old_block_handling,
|
||
|
|
dimension,
|
||
|
|
is_legacy,
|
||
|
|
update_neighbors,
|
||
|
|
)
|
||
|
|
return {
|
||
|
|
"changed": bool(changed),
|
||
|
|
"position": [x, y, z],
|
||
|
|
"dimension": dimension,
|
||
|
|
"block": {"name": name, "aux": aux},
|
||
|
|
"old_block_handling": old_block_handling,
|
||
|
|
"is_legacy": is_legacy,
|
||
|
|
"update_neighbors": update_neighbors,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _place_block(params):
|
||
|
|
request = _require_params(params)
|
||
|
|
task = {
|
||
|
|
"request": request,
|
||
|
|
"done": threading.Event(),
|
||
|
|
"cancelled": False,
|
||
|
|
"started": False,
|
||
|
|
}
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
if not _SERVER_ACTIVE:
|
||
|
|
raise BridgeError("server system is not ready")
|
||
|
|
if len(_SERVER_TASKS) >= MAX_PENDING_TASKS:
|
||
|
|
raise BridgeError("server request queue is full")
|
||
|
|
_SERVER_TASKS.append(task)
|
||
|
|
|
||
|
|
task["done"].wait(5.0)
|
||
|
|
if not _event_is_set(task["done"]):
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
if task["started"]:
|
||
|
|
started = True
|
||
|
|
else:
|
||
|
|
started = False
|
||
|
|
task["cancelled"] = True
|
||
|
|
if task in _SERVER_TASKS:
|
||
|
|
_SERVER_TASKS.remove(task)
|
||
|
|
if started:
|
||
|
|
task["done"].wait()
|
||
|
|
else:
|
||
|
|
raise BridgeError("place_block timed out waiting for server thread")
|
||
|
|
if "error" in task:
|
||
|
|
raise BridgeError(task["error"])
|
||
|
|
return task["result"]
|
||
|
|
|
||
|
|
|
||
|
|
def UPDATE():
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
tasks = list(_SERVER_TASKS)
|
||
|
|
del _SERVER_TASKS[:]
|
||
|
|
|
||
|
|
for task in tasks:
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
if task["cancelled"]:
|
||
|
|
task["done"].set()
|
||
|
|
continue
|
||
|
|
task["started"] = True
|
||
|
|
try:
|
||
|
|
task["result"] = _place_block_on_server(task["request"])
|
||
|
|
except BridgeError as err:
|
||
|
|
task["error"] = str(err)
|
||
|
|
except Exception as err:
|
||
|
|
traceback.print_exc()
|
||
|
|
task["error"] = str(err)
|
||
|
|
finally:
|
||
|
|
task["done"].set()
|
||
|
|
|
||
|
|
|
||
|
|
def _cancel_server_tasks():
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
tasks = list(_SERVER_TASKS)
|
||
|
|
del _SERVER_TASKS[:]
|
||
|
|
for task in tasks:
|
||
|
|
task["cancelled"] = True
|
||
|
|
task["error"] = "server stopped before executing request"
|
||
|
|
task["done"].set()
|
||
|
|
|
||
|
|
|
||
|
|
class MCPBridge(object):
|
||
|
|
def __init__(self, state_path):
|
||
|
|
self.state_path = state_path
|
||
|
|
self.sock = None
|
||
|
|
self.lock = threading.Lock()
|
||
|
|
self.running = False
|
||
|
|
self.connections = set()
|
||
|
|
self.accept_thread = None
|
||
|
|
self.client_threads = set()
|
||
|
|
self.port = None
|
||
|
|
self.tmp_path = None
|
||
|
|
|
||
|
|
def start(self):
|
||
|
|
with self.lock:
|
||
|
|
if self.running:
|
||
|
|
return
|
||
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
|
|
sock.bind(("127.0.0.1", 0))
|
||
|
|
sock.listen(4)
|
||
|
|
sock.settimeout(0.2)
|
||
|
|
self.sock = sock
|
||
|
|
self.running = True
|
||
|
|
port = sock.getsockname()[1]
|
||
|
|
self.port = port
|
||
|
|
|
||
|
|
self._write_state(port)
|
||
|
|
print("[MCPBridge] Listening on 127.0.0.1:%d" % port)
|
||
|
|
thread = _create_daemon_thread(self._accept_loop)
|
||
|
|
with self.lock:
|
||
|
|
self.accept_thread = thread
|
||
|
|
thread.start()
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
with self.lock:
|
||
|
|
sock = self.sock
|
||
|
|
self.sock = None
|
||
|
|
self.running = False
|
||
|
|
connections = list(self.connections)
|
||
|
|
self.connections.clear()
|
||
|
|
accept_thread = self.accept_thread
|
||
|
|
self.accept_thread = None
|
||
|
|
client_threads = list(self.client_threads)
|
||
|
|
self.client_threads.clear()
|
||
|
|
if sock:
|
||
|
|
try:
|
||
|
|
sock.shutdown(socket.SHUT_RDWR)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
sock.close()
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
for conn in connections:
|
||
|
|
try:
|
||
|
|
conn.shutdown(socket.SHUT_RDWR)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
conn.close()
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
if accept_thread and accept_thread.is_alive():
|
||
|
|
accept_thread.join(1.0)
|
||
|
|
for thread in client_threads:
|
||
|
|
if thread.is_alive():
|
||
|
|
thread.join(1.0)
|
||
|
|
self._delete_state()
|
||
|
|
|
||
|
|
def _write_state(self, port):
|
||
|
|
state_dir = os.path.dirname(self.state_path)
|
||
|
|
if state_dir and not os.path.isdir(state_dir):
|
||
|
|
os.makedirs(state_dir)
|
||
|
|
tmp_path = "%s.%d.tmp" % (self.state_path, port)
|
||
|
|
self.tmp_path = tmp_path
|
||
|
|
data = json.dumps(
|
||
|
|
{"version": 1, "host": "127.0.0.1", "port": port},
|
||
|
|
separators=(",", ":"),
|
||
|
|
)
|
||
|
|
with open(tmp_path, "w") as fp:
|
||
|
|
fp.write(data)
|
||
|
|
os.rename(tmp_path, self.state_path)
|
||
|
|
self.tmp_path = None
|
||
|
|
|
||
|
|
def _delete_state(self):
|
||
|
|
try:
|
||
|
|
if self.tmp_path and os.path.exists(self.tmp_path):
|
||
|
|
os.remove(self.tmp_path)
|
||
|
|
self.tmp_path = None
|
||
|
|
if not os.path.exists(self.state_path) or self.port is None:
|
||
|
|
return
|
||
|
|
with open(self.state_path, "r") as fp:
|
||
|
|
state = json.load(fp)
|
||
|
|
if state.get("host") == "127.0.0.1" and state.get("port") == self.port:
|
||
|
|
os.remove(self.state_path)
|
||
|
|
except Exception:
|
||
|
|
traceback.print_exc()
|
||
|
|
finally:
|
||
|
|
self.port = None
|
||
|
|
|
||
|
|
def _accept_loop(self):
|
||
|
|
while True:
|
||
|
|
with self.lock:
|
||
|
|
sock = self.sock
|
||
|
|
running = self.running
|
||
|
|
if not running or sock is None:
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
conn, _addr = sock.accept()
|
||
|
|
except socket.timeout:
|
||
|
|
continue
|
||
|
|
except socket.error:
|
||
|
|
with self.lock:
|
||
|
|
if not self.running:
|
||
|
|
return
|
||
|
|
traceback.print_exc()
|
||
|
|
return
|
||
|
|
except Exception:
|
||
|
|
with self.lock:
|
||
|
|
if not self.running:
|
||
|
|
return
|
||
|
|
traceback.print_exc()
|
||
|
|
return
|
||
|
|
with self.lock:
|
||
|
|
if not self.running:
|
||
|
|
conn.close()
|
||
|
|
return
|
||
|
|
self.connections.add(conn)
|
||
|
|
thread = _create_daemon_thread(self._handle_client, (conn,))
|
||
|
|
self.client_threads.add(thread)
|
||
|
|
conn.settimeout(0.2)
|
||
|
|
thread.start()
|
||
|
|
|
||
|
|
def _handle_client(self, conn):
|
||
|
|
try:
|
||
|
|
pending = b""
|
||
|
|
while True:
|
||
|
|
with self.lock:
|
||
|
|
if not self.running:
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
chunk = conn.recv(4096)
|
||
|
|
except socket.timeout:
|
||
|
|
continue
|
||
|
|
if not chunk:
|
||
|
|
return
|
||
|
|
pending += chunk
|
||
|
|
if len(pending) > MAX_REQUEST_BYTES:
|
||
|
|
raise BridgeError("request line exceeds maximum size")
|
||
|
|
while b"\n" in pending:
|
||
|
|
raw_line, pending = pending.split(b"\n", 1)
|
||
|
|
line = _to_text(raw_line).strip()
|
||
|
|
if line:
|
||
|
|
self._handle_line(conn, line)
|
||
|
|
except Exception:
|
||
|
|
with self.lock:
|
||
|
|
if self.running:
|
||
|
|
traceback.print_exc()
|
||
|
|
finally:
|
||
|
|
with self.lock:
|
||
|
|
self.connections.discard(conn)
|
||
|
|
self.client_threads.discard(threading.currentThread())
|
||
|
|
try:
|
||
|
|
conn.close()
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
def _handle_line(self, conn, line):
|
||
|
|
request_id = None
|
||
|
|
try:
|
||
|
|
request = json.loads(line)
|
||
|
|
if not isinstance(request, dict):
|
||
|
|
raise BridgeError("request must be an object")
|
||
|
|
request_id = request.get("id")
|
||
|
|
method = request.get("method")
|
||
|
|
if method != "place_block":
|
||
|
|
raise BridgeError("unknown MCP bridge method: %s" % method)
|
||
|
|
result = _place_block(request.get("params", {}))
|
||
|
|
_send_json_line(conn, {"id": request_id, "ok": True, "result": result})
|
||
|
|
except BridgeError as err:
|
||
|
|
_send_json_line(conn, {"id": request_id, "ok": False, "error": str(err)})
|
||
|
|
except Exception as err:
|
||
|
|
traceback.print_exc()
|
||
|
|
_send_json_line(conn, {"id": request_id, "ok": False, "error": str(err)})
|
||
|
|
|
||
|
|
|
||
|
|
def ON_SERVER_INIT():
|
||
|
|
global _BRIDGE, _SERVER_ACTIVE
|
||
|
|
state_path = GET_MCP_BRIDGE_STATE_PATH()
|
||
|
|
if not state_path:
|
||
|
|
return
|
||
|
|
if _BRIDGE:
|
||
|
|
return
|
||
|
|
bridge = MCPBridge(state_path)
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
_SERVER_ACTIVE = True
|
||
|
|
try:
|
||
|
|
bridge.start()
|
||
|
|
except Exception:
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
_SERVER_ACTIVE = False
|
||
|
|
bridge.close()
|
||
|
|
raise
|
||
|
|
_BRIDGE = bridge
|
||
|
|
|
||
|
|
|
||
|
|
def ON_SERVER_EXIT():
|
||
|
|
global _BRIDGE, _SERVER_ACTIVE
|
||
|
|
bridge = _BRIDGE
|
||
|
|
_BRIDGE = None
|
||
|
|
with _SERVER_TASKS_LOCK:
|
||
|
|
_SERVER_ACTIVE = False
|
||
|
|
_cancel_server_tasks()
|
||
|
|
if bridge:
|
||
|
|
bridge.close()
|