def run(self):
while True:<EOL><INDENT>try:<EOL><INDENT>doc = self.in_queue.get(True, QUEUE_TIMEOUT)<EOL>if doc == UDF.QUEUE_CLOSED:<EOL><INDENT>self.in_queue.put(UDF.QUEUE_CLOSED)<EOL>break<EOL><DEDENT>self.session.add_all(y for y in self.apply(doc, **self.apply_kwargs))<EOL>self.out_queue.put(UDF.TASK_DONE)<EOL><DEDENT>except Empty:<EOL><INDENT>continue<EOL><DEDENT><DEDENT>self.session.commit()<EOL>self.session.close()<EOL>
This method is called when the UDF is run as a Process in a
multiprocess setting The basic routine is: get from JoinableQueue,
apply, put / add outputs, loop