implement async abstract methods, add TCP conn

projects/jdwp/BUCK

@@ -3,12 +3,5 @@
 python_binary(
   name = "main",
   main = "main.py",
-  deps = [],
-)
-
-
-python_library(
-    name = "lib",
-    srcs = glob(["**/*.py"]),
-    visibility = ["PUBLIC"],
+  deps = ["//projects/jdwp/runtime:runtime"],
 )

projects/jdwp/main.py

@@ -1,9 +1,21 @@
 # Copyright (c) Meta Platforms, Inc. and affiliates.
 
+import asyncio
+from projects.jdwp.runtime.jvm_connection import JVMConnection
 
-def main():
-    return None
+
+async def main():
+    host = "localhost"
+    port = 8880
+
+    # connection = JVMConnection(host, port)
+
+    # await connection.connect()
+
+    # await connection.handshake()
+
+    # await connection.close()
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())

projects/jdwp/runtime/BUCK

@@ -19,6 +19,8 @@ python_library(
     ":type-aliases",
     "async_streams.py",
     "jdwpstruct.py",
+    "jdwp_streams.py",
+    "jvm_connection.py",
   ],
   visibility = ["PUBLIC", ],
   deps = [],

projects/jdwp/runtime/jdwp_streams.py

@@ -0,0 +1,203 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+
+import struct
+import typing
+import asyncio
+from projects.jdwp.runtime.type_aliases import *
+from projects.jdwp.runtime.async_streams import (
+    JDWPInputStreamBase,
+    JDWPOutputStreamBase,
+)
+
+class JDWPInputStreamBuffer(JDWPInputStreamBase):
+    def __init__(self, stream_reader: asyncio.StreamReader, size: int):
+        super().__init__()
+        self.__tcp_connection = stream_reader
+        self.__size = size
+
+    async def read_bytes(self, size: int) -> bytes:
+        try:
+            return await self.__tcp_connection.readexactly(size)
+        except Exception as e:
+            print(f"Error during data receiving: {e}")
+            return b""
+
+
+class JDWPInputStream(JDWPInputStreamBase):
+
+    def __init__(self):
+        super().__init__()
+        self.__buffer = JDWPInputStreamBuffer()
+
+    async def read_boolean(self) -> bool:
+        data = await self.__buffer.read_bytes(1)
+        return bool(data[0])
+
+    async def read_location(self) -> typing.Any:
+        pass
+
+    async def read_string(self) -> str:
+        length = await self.read_int()
+        string_data = await self.__buffer.read_bytes(length)
+        return string_data.decode("utf-8")
+
+    async def read_object_id(self) -> ObjectIDType:
+        return ObjectIDType(0)
+
+    async def read_thread_id(self) -> ThreadIDType:
+        return ThreadIDType(0)
+
+    async def read_thread_group_id(self) -> ThreadGroupIDType:
+        return ThreadGroupIDType(0)
+
+    async def read_string_id(self) -> StringIDType:
+        return StringIDType(0)
+
+    async def read_class_loader_id(self) -> ClassLoaderIDType:
+        return ClassLoaderIDType(0)
+
+    async def read_class_object_id(self) -> ClassObjectIDType:
+        return ClassObjectIDType(0)
+
+    async def read_array_id(self) -> ArrayIDType:
+        return ArrayIDType(0)
+
+    async def read_reference_type_id(self) -> ReferenceTypeIDType:
+        return ReferenceTypeIDType(0)
+
+    async def read_class_id(self) -> ClassIDType:
+        return ClassIDType(0)
+
+    async def read_interface_id(self) -> InterfaceIDType:
+        return InterfaceIDType(0)
+
+    async def read_array_type_id(self) -> ArrayTypeIDType:
+        return ArrayTypeIDType(0)
+
+    async def read_method_id(self) -> MethodIDType:
+        return MethodIDType(0)
+
+    async def read_field_id(self) -> FieldIDType:
+        return FieldIDType(0)
+
+    async def read_frame_id(self) -> FrameIDType:
+        return FrameIDType(0)
+
+    async def read_byte(self) -> int:
+        data = await self.__buffer.read_bytes(1)
+        return int.from_bytes(data, byteorder="big")
+
+    async def read_int(self) -> int:
+        data = await self.__buffer.read_bytes(4)
+        return struct.unpack("!I", data)[0]
+
+    async def read_long(self) -> int:
+        data = await self.__buffer.read_bytes(8)
+        return struct.unpack("!Q", data)[0]
+
+
+class JDWPOutputStreamBuffer(JDWPOutputStreamBase):
+    def __init__(self):
+        super().__init__()
+        self.__buffer = bytes()
+        self.__size = 0
+
+    def get_buffer_size(self) -> int:
+        return self.__size
+
+    def get_buffer(self) -> bytes:
+        return self.__buffer
+
+    async def flush(self, target_stream: asyncio.StreamWriter) -> None:
+        message_size = self.__size
+        header_bytes = struct.pack("!I", message_size)
+
+        await target_stream.write(header_bytes)
+        await target_stream.drain()
+
+        await target_stream.write(self.__buffer)
+        await target_stream.drain()
+        self.__buffer = bytes()
+        self.__size = 0
+
+    def write(self, data: bytes) -> None:
+        self.__buffer += data
+        self.__size += len(data)
+
+
+class JDWPOutputStream(JDWPOutputStreamBase):
+    def __init__(self):
+        super().__init__()
+        self.__buffer = JDWPOutputStreamBuffer()
+
+    def write_boolean(self, value: bool) -> None:
+        self._write_bytes(struct.pack("!B", int(value)))
+
+    def write_int(self, value: int) -> None:
+        self._write_bytes(struct.pack("!I", value))
+
+    def write_array_id(self, value: ArrayIDType) -> None:
+        self._write_id(value)
+
+    def write_array_type_id(self, value: ArrayTypeIDType) -> None:
+        self._write_id(value)
+
+    def write_byte(self, value: int) -> None:
+        self._write_bytes(struct.pack("!B", value))
+
+    def write_class_id(self, value: ClassIDType) -> None:
+        self._write_id(value)
+
+    def write_class_loader_id(self, value: ClassLoaderIDType) -> None:
+        self._write_id(value)
+
+    def write_class_object_id(self, value: ClassObjectIDType) -> None:
+        self._write_id(value)
+
+    def write_field_id(self, value: FieldIDType) -> None:
+        self._write_id(value)
+
+    def write_frame_id(self, value: FrameIDType) -> None:
+        self._write_id(value)
+
+    def write_interface_id(self, value: InterfaceIDType) -> None:
+        self._write_id(value)
+
+    def write_location(self, value: typing.Any) -> None:
+        pass
+
+    def write_method_id(self, value: MethodIDType) -> None:
+        self._write_id(value)
+
+    def write_string_id(self, value: StringIDType) -> None:
+        self._write_id(value)
+
+    def write_thread_group_id(self, value: ThreadGroupIDType) -> None:
+        self._write_id(value)
+
+    def write_long(self, value: int) -> None:
+        self._write_bytes(struct.pack("!Q", value))
+
+    def write_object_id(self, value: typing.Any) -> None:
+        self._write_id(value)
+
+    def write_thread_id(self, value: typing.Any) -> None:
+        self._write_id(value)
+
+    def write_string(self, value: str) -> None:
+        value_bytes = value.encode("utf-8")
+        length = len(value_bytes)
+
+        self.write_int(length)
+
+        self._write_bytes(value_bytes)
+
+    def write_reference_type_id(self, value: ReferenceTypeIDType) -> None:
+        self._write_id(value)
+
+    def _write_bytes(self, data: bytes):
+        self.__buffer.write(data)
+
+    def _write_id(self, value: typing.Any) -> None:
+        self._write_bytes(struct.pack("B", value))
+        self._write_bytes(value.to_bytes(value, byteorder="big"))

projects/jdwp/runtime/jvm_connection.py

@@ -0,0 +1,119 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+
+import asyncio
+import typing
+import struct
+from projects.jdwp.runtime.jdwp_streams import JDWPInputStream, JDWPOutputStream, JDWPInputStreamBuffer, JDWPOutputStreamBuffer
+
+
+class PacketHeader(typing.NamedTuple):
+    length: int
+    id: int
+    flags: bytes
+    command_set: int
+    command: int
+
+    def write(self) -> bytes:
+        return struct.pack(
+            "!IIcBB", self.length, self.id, self.flags, self.command_set, self.command
+        )
+
+
+class JVMConnection:
+    def __init__(self, host: str, port: int):
+        self.host: str = host
+        self.port: int = port
+        self.next_packet_id: int = 0
+        self.__reader: typing.Optional[asyncio.StreamReader] = None
+        self.__writer: typing.Optional[asyncio.StreamWriter] = None
+        self.__input_stream: typing.Optional[JDWPInputStreamBuffer] = None
+        self.__output_stream: typing.Optional[JDWPOutputStreamBuffer] = None
+
+    async def connect(self) -> None:
+        try:
+            self.__reader, self.__writer = await asyncio.open_connection(
+                self.host, self.port
+            )
+
+            if self.__writer is None:
+                raise Exception("Stream writer not initialized")
+            self.__output_stream = JDWPOutputStreamBuffer(stream_writer=self.__writer, size=0)
+            if self.__reader is None:
+                raise Exception("Stream reader not initialized")
+            self.__input_stream = JDWPInputStreamBuffer(stream_reader=self.__reader,  size=0)
+
+        except Exception as e:
+            print(f"Error during connection: {e}")
+            exit(1)
+
+    def close(self) -> None:
+        if self.__writer is not None:
+            self.__writer.close()
+        self.__input_stream, self.__output_stream = None, None
+
+    async def handshake(self) -> None:
+        try:
+            handshake_bytes = b"JDWP-Handshake"
+            if self.__output_stream is None:
+                raise Exception("Output stream not initialized")
+            __stream = self.__output_stream
+            __stream.write(handshake_bytes)
+            if self.__writer is None:
+                raise Exception("Stream writer not initialized")
+            await __stream.flush(self.__writer)
+
+            if self.__input_stream is None:
+                raise Exception("Input stream not initialized")
+            response_bytes = await self.__input_stream.read_bytes(len(handshake_bytes))
+
+            if response_bytes != handshake_bytes:
+                raise Exception("Invalid handshake response")
+
+            print(f"Handshake successful")
+        except Exception as e:
+            print(f"Error during handshake: {e}")
+            self.close()
+
+    async def __read_packet_header(self) -> PacketHeader:
+        if self.__input_stream is None:
+            raise Exception("Input stream not initialized")
+
+        header_data = await self.__input_stream.read_bytes(10)
+        return PacketHeader(*struct.unpack("!IIcBB", header_data))
+
+    async def __read_packet_data(self, length: int) -> bytes:
+        if self.__input_stream is None:
+            raise Exception("Input stream not initialized")
+        return await self.__input_stream.read_bytes(length)
+
+    async def __write_packet_header(
+        self, length: int, flags: bytes, command_set: int, command: int
+    ) -> None:
+        try:
+            packet_id = self.__get_next_packet_id()
+            packet_header = PacketHeader(length, packet_id, flags, command_set, command)
+            header_data = packet_header.write()
+            if self.__output_stream is None:
+                raise Exception("Output stream not initialized")
+            __stream = self.__output_stream
+            __stream.write(header_data)
+            if self.__writer is None:
+                raise Exception("Stream writer not initialized")
+            await __stream.flush(self.__writer)
+
+        except Exception as e:
+            print(f"Error writing packet header: {e}")
+            self.close()
+
+    async def __write_packet_data(self, data: bytes) -> None:
+        if self.__output_stream is None:
+                raise Exception("Output stream not initialized")
+        __stream = self.__output_stream
+        __stream.write(data)
+        if self.__writer is None:
+                raise Exception("Stream writer not initialized")
+        await __stream.flush(self.__writer)
+
+    def __get_next_packet_id(self) -> int:
+        self.next_packet_id += 1
+        return self.next_packet_id