<s> from __future__ import absolute_import <EOL> from itertools import cycle <EOL> import logging <EOL> import random <EOL> from six . moves import xrange <EOL> from . base import Producer <EOL> log = logging . getLogger ( __name__ ) <EOL> class SimpleProducer ( Producer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . partition_cycles = { } <EOL> self . random_start = kwargs . pop ( '<STR_LIT>' , True ) <EOL> super ( SimpleProducer , self ) . __init__ ( * args , ** kwargs ) <EOL> def _next_partition ( self , topic ) : <EOL> if topic not in self . partition_cycles : <EOL> if not self . client . has_metadata_for_topic ( topic ) : <EOL> self . client . ensure_topic_exists ( topic ) <EOL> self . partition_cycles [ topic ] = cycle ( self . client . get_partition_ids_for_topic ( topic ) ) <EOL> if self . random_start : <EOL> num_partitions = len ( self . client . get_partition_ids_for_topic ( topic ) ) <EOL> for _ in xrange ( random . randint ( <NUM_LIT:0> , num_partitions - <NUM_LIT:1> ) ) : <EOL> next ( self . partition_cycles [ topic ] ) <EOL> return next ( self . partition_cycles [ topic ] ) <EOL> def send_messages ( self , topic , * msg ) : <EOL> partition = self . _next_partition ( topic ) <EOL> return super ( SimpleProducer , self ) . send_messages ( <EOL> topic , partition , * msg <EOL> ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . async </s>
