| |
| <h1>db_send.py</h1> |
| <div class="highlight"><pre><span></span><span class="kn">from</span> <span class="nn">__future__</span> <span class="kn">import</span> <span class="n">print_function</span><span class="p">,</span> <span class="n">unicode_literals</span> |
| <span class="kn">import</span> <span class="nn">optparse</span> |
| <span class="kn">import</span> <span class="nn">time</span> |
| <span class="k">try</span><span class="p">:</span> |
| <span class="kn">import</span> <span class="nn">Queue</span> |
| <span class="k">except</span> <span class="ne">ImportError</span><span class="p">:</span> |
| <span class="kn">import</span> <span class="nn">queue</span> <span class="k">as</span> <span class="nn">Queue</span> |
| |
| |
| <span class="kn">from</span> <span class="nn">proton</span> <span class="kn">import</span> <span class="n">Message</span> |
| <span class="kn">from</span> <span class="nn">proton.handlers</span> <span class="kn">import</span> <span class="n">MessagingHandler</span> |
| <span class="kn">from</span> <span class="nn">proton.reactor</span> <span class="kn">import</span> <span class="n">ApplicationEvent</span><span class="p">,</span> <span class="n">Container</span><span class="p">,</span> <span class="n">EventInjector</span> |
| <span class="kn">from</span> <span class="nn">db_common</span> <span class="kn">import</span> <span class="n">Db</span> |
| |
| |
| <span class="k">class</span> <span class="nc">Send</span><span class="p">(</span><span class="n">MessagingHandler</span><span class="p">):</span> |
| <span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">url</span><span class="p">,</span> <span class="n">count</span><span class="p">):</span> |
| <span class="nb">super</span><span class="p">(</span><span class="n">Send</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="fm">__init__</span><span class="p">()</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">url</span> <span class="o">=</span> <span class="n">url</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">delay</span> <span class="o">=</span> <span class="mi">0</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">sent</span> <span class="o">=</span> <span class="mi">0</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">confirmed</span> <span class="o">=</span> <span class="mi">0</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">load_count</span> <span class="o">=</span> <span class="mi">0</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">records</span> <span class="o">=</span> <span class="n">Queue</span><span class="o">.</span><span class="n">Queue</span><span class="p">(</span><span class="n">maxsize</span><span class="o">=</span><span class="mi">50</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">target</span> <span class="o">=</span> <span class="n">count</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">db</span> <span class="o">=</span> <span class="n">Db</span><span class="p">(</span><span class="s2">"src_db"</span><span class="p">,</span> <span class="n">EventInjector</span><span class="p">())</span> |
| |
| <span class="k">def</span> <span class="nf">keep_sending</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span> |
| <span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">target</span> <span class="o">==</span> <span class="mi">0</span> <span class="ow">or</span> <span class="bp">self</span><span class="o">.</span><span class="n">sent</span> <span class="o"><</span> <span class="bp">self</span><span class="o">.</span><span class="n">target</span> |
| |
| <span class="k">def</span> <span class="nf">on_start</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">container</span> <span class="o">=</span> <span class="n">event</span><span class="o">.</span><span class="n">container</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">container</span><span class="o">.</span><span class="n">selectable</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">db</span><span class="o">.</span><span class="n">injector</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">sender</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">container</span><span class="o">.</span><span class="n">create_sender</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">url</span><span class="p">)</span> |
| |
| <span class="k">def</span> <span class="nf">on_records_loaded</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">records</span><span class="o">.</span><span class="n">empty</span><span class="p">():</span> |
| <span class="k">if</span> <span class="n">event</span><span class="o">.</span><span class="n">subject</span> <span class="o">==</span> <span class="bp">self</span><span class="o">.</span><span class="n">load_count</span><span class="p">:</span> |
| <span class="nb">print</span><span class="p">(</span><span class="s2">"Exhausted available data, waiting to recheck..."</span><span class="p">)</span> |
| <span class="c1"># check for new data after 5 seconds</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">container</span><span class="o">.</span><span class="n">schedule</span><span class="p">(</span><span class="mi">5</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span> |
| <span class="k">else</span><span class="p">:</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">send</span><span class="p">()</span> |
| |
| <span class="k">def</span> <span class="nf">request_records</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span> |
| <span class="k">if</span> <span class="ow">not</span> <span class="bp">self</span><span class="o">.</span><span class="n">records</span><span class="o">.</span><span class="n">full</span><span class="p">():</span> |
| <span class="nb">print</span><span class="p">(</span><span class="s2">"loading records..."</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">load_count</span> <span class="o">+=</span> <span class="mi">1</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">db</span><span class="o">.</span><span class="n">load</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">records</span><span class="p">,</span> <span class="n">event</span><span class="o">=</span><span class="n">ApplicationEvent</span><span class="p">(</span> |
| <span class="s2">"records_loaded"</span><span class="p">,</span> <span class="n">link</span><span class="o">=</span><span class="bp">self</span><span class="o">.</span><span class="n">sender</span><span class="p">,</span> <span class="n">subject</span><span class="o">=</span><span class="bp">self</span><span class="o">.</span><span class="n">load_count</span><span class="p">))</span> |
| |
| <span class="k">def</span> <span class="nf">on_sendable</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">send</span><span class="p">()</span> |
| |
| <span class="k">def</span> <span class="nf">send</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span> |
| <span class="k">while</span> <span class="bp">self</span><span class="o">.</span><span class="n">sender</span><span class="o">.</span><span class="n">credit</span> <span class="ow">and</span> <span class="ow">not</span> <span class="bp">self</span><span class="o">.</span><span class="n">records</span><span class="o">.</span><span class="n">empty</span><span class="p">():</span> |
| <span class="k">if</span> <span class="ow">not</span> <span class="bp">self</span><span class="o">.</span><span class="n">keep_sending</span><span class="p">():</span> |
| <span class="k">return</span> |
| <span class="n">record</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">records</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="kc">False</span><span class="p">)</span> |
| <span class="nb">id</span> <span class="o">=</span> <span class="n">record</span><span class="p">[</span><span class="s1">'id'</span><span class="p">]</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">sender</span><span class="o">.</span><span class="n">send</span><span class="p">(</span><span class="n">Message</span><span class="p">(</span><span class="nb">id</span><span class="o">=</span><span class="nb">id</span><span class="p">,</span> <span class="n">durable</span><span class="o">=</span><span class="kc">True</span><span class="p">,</span> <span class="n">body</span><span class="o">=</span><span class="n">record</span><span class="p">[</span><span class="s1">'description'</span><span class="p">]),</span> <span class="n">tag</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="nb">id</span><span class="p">))</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">sent</span> <span class="o">+=</span> <span class="mi">1</span> |
| <span class="nb">print</span><span class="p">(</span><span class="s2">"sent message </span><span class="si">%s</span><span class="s2">"</span> <span class="o">%</span> <span class="nb">id</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">request_records</span><span class="p">()</span> |
| |
| <span class="k">def</span> <span class="nf">on_settled</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="nb">id</span> <span class="o">=</span> <span class="nb">int</span><span class="p">(</span><span class="n">event</span><span class="o">.</span><span class="n">delivery</span><span class="o">.</span><span class="n">tag</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">db</span><span class="o">.</span><span class="n">delete</span><span class="p">(</span><span class="nb">id</span><span class="p">)</span> |
| <span class="nb">print</span><span class="p">(</span><span class="s2">"settled message </span><span class="si">%s</span><span class="s2">"</span> <span class="o">%</span> <span class="nb">id</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">confirmed</span> <span class="o">+=</span> <span class="mi">1</span> |
| <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">confirmed</span> <span class="o">==</span> <span class="bp">self</span><span class="o">.</span><span class="n">target</span><span class="p">:</span> |
| <span class="n">event</span><span class="o">.</span><span class="n">connection</span><span class="o">.</span><span class="n">close</span><span class="p">()</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">db</span><span class="o">.</span><span class="n">close</span><span class="p">()</span> |
| |
| <span class="k">def</span> <span class="nf">on_disconnected</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">db</span><span class="o">.</span><span class="n">reset</span><span class="p">()</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">sent</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">confirmed</span> |
| |
| <span class="k">def</span> <span class="nf">on_timer_task</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">event</span><span class="p">):</span> |
| <span class="nb">print</span><span class="p">(</span><span class="s2">"Rechecking for data..."</span><span class="p">)</span> |
| <span class="bp">self</span><span class="o">.</span><span class="n">request_records</span><span class="p">()</span> |
| |
| |
| <span class="n">parser</span> <span class="o">=</span> <span class="n">optparse</span><span class="o">.</span><span class="n">OptionParser</span><span class="p">(</span><span class="n">usage</span><span class="o">=</span><span class="s2">"usage: %prog [options]"</span><span class="p">,</span> |
| <span class="n">description</span><span class="o">=</span><span class="s2">"Send messages to the supplied address."</span><span class="p">)</span> |
| <span class="n">parser</span><span class="o">.</span><span class="n">add_option</span><span class="p">(</span><span class="s2">"-a"</span><span class="p">,</span> <span class="s2">"--address"</span><span class="p">,</span> <span class="n">default</span><span class="o">=</span><span class="s2">"localhost:5672/examples"</span><span class="p">,</span> |
| <span class="n">help</span><span class="o">=</span><span class="s2">"address to which messages are sent (default </span><span class="si">%d</span><span class="s2">efault)"</span><span class="p">)</span> |
| <span class="n">parser</span><span class="o">.</span><span class="n">add_option</span><span class="p">(</span><span class="s2">"-m"</span><span class="p">,</span> <span class="s2">"--messages"</span><span class="p">,</span> <span class="nb">type</span><span class="o">=</span><span class="s2">"int"</span><span class="p">,</span> <span class="n">default</span><span class="o">=</span><span class="mi">0</span><span class="p">,</span> |
| <span class="n">help</span><span class="o">=</span><span class="s2">"number of messages to send; 0 sends indefinitely (default </span><span class="si">%d</span><span class="s2">efault)"</span><span class="p">)</span> |
| <span class="n">opts</span><span class="p">,</span> <span class="n">args</span> <span class="o">=</span> <span class="n">parser</span><span class="o">.</span><span class="n">parse_args</span><span class="p">()</span> |
| |
| <span class="k">try</span><span class="p">:</span> |
| <span class="n">Container</span><span class="p">(</span><span class="n">Send</span><span class="p">(</span><span class="n">opts</span><span class="o">.</span><span class="n">address</span><span class="p">,</span> <span class="n">opts</span><span class="o">.</span><span class="n">messages</span><span class="p">))</span><span class="o">.</span><span class="n">run</span><span class="p">()</span> |
| <span class="k">except</span> <span class="ne">KeyboardInterrupt</span><span class="p">:</span> |
| <span class="k">pass</span> |
| </pre></div> |
| |
| <p><a href="db_send.py">Download this file</a></p> |