Move ConsumerProtocol definition to kafka.coordinator.protocol

1 files changed, 10 insertions, 34 deletions

diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 48d5e14..af3e019 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -1,3 +1,5 @@
+from __future__ import absolute_import
+
 import copy
 import collections
 import logging
@@ -6,44 +8,18 @@ import time
 import six
 
 from .base import BaseCoordinator
-import kafka.common as Errors
-from kafka.common import OffsetAndMetadata, TopicPartition
-from kafka.future import Future
-from kafka.protocol.commit import (
+from .protocol import (
+    ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment,
+    ConsumerProtocol)
+from ..common import OffsetAndMetadata, TopicPartition
+from ..future import Future
+from ..protocol.commit import (
     OffsetCommitRequest_v2, OffsetCommitRequest_v1, OffsetCommitRequest_v0,
     OffsetFetchRequest_v0, OffsetFetchRequest_v1)
-from kafka.protocol.struct import Struct
-from kafka.protocol.types import Array, Bytes, Int16, Int32, Schema, String
-
-log = logging.getLogger(__name__)
-
-
-class ConsumerProtocolMemberMetadata(Struct):
-    SCHEMA = Schema(
-        ('version', Int16),
-        ('subscription', Array(String('utf-8'))),
-        ('user_data', Bytes))
-
-
-class ConsumerProtocolMemberAssignment(Struct):
-    SCHEMA = Schema(
-        ('version', Int16),
-        ('assignment', Array(
-            ('topic', String('utf-8')),
-            ('partitions', Array(Int32)))),
-        ('user_data', Bytes))
-
-    def partitions(self):
-        return [TopicPartition(topic, partition)
-                for topic, partitions in self.assignment # pylint: disable-msg=no-member
-                for partition in partitions]
 
+import kafka.common as Errors
 
-class ConsumerProtocol(object):
-    PROTOCOL_TYPE = 'consumer'
-    ASSIGNMENT_STRATEGIES = ('roundrobin',)
-    METADATA = ConsumerProtocolMemberMetadata
-    ASSIGNMENT = ConsumerProtocolMemberAssignment
+log = logging.getLogger(__name__)
 
 
 class ConsumerCoordinator(BaseCoordinator):