#NEWLINE# def run( publish , consume , consumers_count = 1 ) : #NEWLINE# #INDENT# 'Run broker.\n\n    publish() put to queue, consume() process one element from queue.\n\n    When publish() is finished and elements from queue are processed process\n    is finished all consumers threads are cleaned.\n\n    :param publish: Function that puts values to the queue\n    :param consume: Function that processes a single value from the queue\n    :param consumers_count: Number of consumers\n    ' #NEWLINE# queue = collections . deque ( ) #NEWLINE# _publisher ( publish , queue ) #NEWLINE# consumers = [ ] #NEWLINE# for i in range ( consumers_count ) : #NEWLINE# #INDENT# consumer = threading . Thread ( target = _consumer , args = ( consume , queue ) ) #NEWLINE# consumer . start ( ) #NEWLINE# consumers . append ( queue ) #NEWLINE# #UNINDENT# for consumer in consumers : #NEWLINE# #INDENT# consumer . join ( )