Broadcast Generators.ipynb

{
 "metadata": {
  "name": ""
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def follow(filename):\n",
      "    with open(filename) as fd:\n",
      "        while True:\n",
      "            line = fd.xreadlines()\n",
      "            yield next(line).strip()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# genmultiplex.py\n",
      "#\n",
      "# Multiplex many different generators into a single real-time stream using threads.\n",
      "\n",
      "import time\n",
      "import threading, Queue\n",
      "\n",
      "#from genqueue import *\n",
      "#from gencat import *\n",
      "\n",
      "def sendto_queue(source, out_q):\n",
      "    for s in source:\n",
      "        out_q.put(s)\n",
      "    \n",
      "def genfrom_queue(in_q):\n",
      "    while True:\n",
      "        try:\n",
      "            value = in_q.get(block=True, timeout=2)\n",
      "        except Queue.Empty:\n",
      "            break\n",
      "        else:\n",
      "            yield value\n",
      "\n",
      "def gen_cat(sources):\n",
      "    for source in sources:\n",
      "        for s in source:\n",
      "            yield s\n",
      "    \n",
      "def multiplex(sources):\n",
      "    in_q = Queue.Queue()\n",
      "    consumers = []\n",
      "    for s in sources:\n",
      "        thr = threading.Thread(target=sendto_queue,\n",
      "                               args=(s,in_q))\n",
      "        thr.start()\n",
      "        consumers.append(genfrom_queue(in_q))\n",
      "    return gen_cat(consumers)\n",
      "\n",
      "if __name__ == '__main__':\n",
      "    #import follow\n",
      "    a_log = follow(\"a.log\")\n",
      "    b_log = follow(\"b.log\")\n",
      "    for line in multiplex([a_log,b_log]):\n",
      "        print line"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "log B - 0\n",
        "log B - 1\n",
        "log B - 2\n",
        "log A - 0\n",
        "log A - 1\n",
        "log B - 3\n",
        "log A - 2\n",
        "log A - 3\n",
        "log A - 4\n",
        "log A - 5\n",
        "log A - 6\n",
        "log B - 4\n",
        "log B - 5\n",
        "log B - 6\n",
        "log B - 7\n",
        "log A - 7\n",
        "log A - 8\n",
        "log B - 8\n",
        "log A - 9\n",
        "log B - 9\n",
        "log A - 10\n",
        "log B - 10\n",
        "log B - 11\n",
        "log B - 12\n",
        "log B - 13\n",
        "log A - 11\n",
        "log A - 12\n",
        "log B - 14\n",
        "log A - 13\n",
        "log A - 14\n",
        "log A - 15\n",
        "log A - 16\n",
        "log A - 17\n",
        "log A - 18\n",
        "log A - 19\n"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# broadcast.py\n",
      "#\n",
      "# Broadcast a generator source to a collection of consumers\n",
      "\n",
      "def broadcast(source, consumers):\n",
      "    for item in source:\n",
      "        for c in consumers:\n",
      "            c.send(item)\n",
      "\n",
      "\n",
      "# Example\n",
      "if __name__ == '__main__':\n",
      "\n",
      "    class Consumer(object):\n",
      "        def send(self,item):\n",
      "            print self, \"got\", item\n",
      "\n",
      "    c1 = Consumer()\n",
      "    c2 = Consumer()\n",
      "    c3 = Consumer()\n",
      "\n",
      "    #from follow import *\n",
      "    lines = follow(\"b.log\")\n",
      "    broadcast(lines,[c1,c2,c3])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}