Rework protocol type definition: AbstractType, Schema, Struct

1 files changed, 30 insertions, 0 deletions

diff --git a/kafka/protocol/fetch.py b/kafka/protocol/fetch.py
new file mode 100644
index 0000000..c6d60cc
--- /dev/null
+++ b/kafka/protocol/fetch.py
@@ -0,0 +1,30 @@
+from .message import MessageSet
+from .struct import Struct
+from .types import Array, Int16, Int32, Int64, Schema, String
+
+
+class FetchRequest(Struct):
+    API_KEY = 1
+    API_VERSION = 0
+    SCHEMA = Schema(
+        ('replica_id', Int32),
+        ('max_wait_time', Int32),
+        ('min_bytes', Int32),
+        ('topics', Array(
+            ('topic', String('utf-8')),
+            ('partitions', Array(
+                ('partition', Int32),
+                ('offset', Int64),
+                ('max_bytes', Int32)))))
+    )
+
+class FetchResponse(Struct):
+    SCHEMA = Schema(
+        ('topics', Array(
+            ('topics', String('utf-8')),
+            ('partitions', Array(
+                ('partition', Int32),
+                ('error_code', Int16),
+                ('highwater_offset', Int64),
+                ('message_set', MessageSet)))))
+    )