From a85e09df89a43de5b659a0fa4ed35bec37c60e04 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sat, 28 Nov 2015 19:41:06 +0800 Subject: Rework protocol type definition: AbstractType, Schema, Struct --- kafka/protocol/fetch.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 kafka/protocol/fetch.py (limited to 'kafka/protocol/fetch.py') 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))))) + ) -- cgit v1.2.1 From 058567912e8d82c1da5e5ead9e30be532573a173 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 29 Nov 2015 10:00:50 +0800 Subject: Add simple BrokerConnection class; add request.RESPONSE_TYPE class vars --- kafka/protocol/fetch.py | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) (limited to 'kafka/protocol/fetch.py') diff --git a/kafka/protocol/fetch.py b/kafka/protocol/fetch.py index c6d60cc..e00c9ab 100644 --- a/kafka/protocol/fetch.py +++ b/kafka/protocol/fetch.py @@ -3,9 +3,22 @@ from .struct import Struct from .types import Array, Int16, Int32, Int64, Schema, String +class FetchResponse(Struct): + SCHEMA = Schema( + ('topics', Array( + ('topics', String('utf-8')), + ('partitions', Array( + ('partition', Int32), + ('error_code', Int16), + ('highwater_offset', Int64), + ('message_set', MessageSet))))) + ) + + class FetchRequest(Struct): API_KEY = 1 API_VERSION = 0 + RESPONSE_TYPE = FetchResponse SCHEMA = Schema( ('replica_id', Int32), ('max_wait_time', Int32), @@ -17,14 +30,3 @@ class FetchRequest(Struct): ('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))))) - ) -- cgit v1.2.1