Better error handling in base consumer implementation.

vumi/blinkenlights/heartbeat/publisher.py

@@ -2,7 +2,7 @@
 
 from twisted.internet.task import LoopingCall
 
-from vumi.service import Publisher
+from vumi.service import Publisher, Exchange
 from vumi.message import Message
 from vumi import log
 
@@ -38,10 +38,11 @@ class HeartBeatPublisher(Publisher):
 
  HEARTBEAT_PERIOD_SECS = 10
 
+ exchange = Exchange(name="vumi.health",
+ exchange_type="direct", durable=True)
+
  def __init__(self, gen_attrs_func):
  self.routing_key = "heartbeat.inbound"
- self.exchange_name = "vumi.health"
- self.durable = True
  self._task = None
  self._gen_attrs_func = gen_attrs_func
 

vumi/blinkenlights/metrics.py

@@ -8,7 +8,7 @@
 from twisted.internet.task import LoopingCall
 from twisted.python import log
 
-from vumi.service import Publisher, Consumer
+from vumi.service import Publisher, Consumer, Exchange
 from vumi.blinkenlights.message20110818 import MetricMessage
 
 import time
@@ -27,10 +27,8 @@ class MetricManager(Publisher):
  :param on_publish:
  Function to call immediately after metrics after published.
  """
- exchange_name = "vumi.metrics"
- exchange_type = "direct"
+ exchange = Exchange("vumi.metrics", exchange_type="direct", durable=True)
  routing_key = "vumi.metrics"
- durable = True
  auto_delete = False
  delivery_mode = 2
 
@@ -272,10 +270,8 @@ class MetricsConsumer(Consumer):
  aggregator (list of aggregator names) and values (a
  list of timestamp and value paits).
  """
- exchange_name = "vumi.metrics"
- exchange_type = "direct"
+ exchange = Exchange("vumi.metrics", exchange_type="direct", durable=True)
  routing_key = "vumi.metrics"
- durable = True
 
  def __init__(self, callback):
  self.callback = callback

vumi/blinkenlights/metrics_workers.py

@@ -11,7 +11,7 @@
 from twisted.internet.task import LoopingCall
 from twisted.internet.protocol import DatagramProtocol
 
-from vumi.service import Consumer, Publisher, Worker
+from vumi.service import Consumer, Publisher, Worker, Exchange
 from vumi.blinkenlights.metrics import (MetricsConsumer, MetricManager, Count,
  Metric, Timer, Aggregator)
 from vumi.blinkenlights.message20110818 import MetricMessage
@@ -27,9 +27,8 @@ class AggregatedMetricConsumer(Consumer):
  parameters are metric_name (str) and values (a list of
  timestamp and value pairs).
  """
- exchange_name = "vumi.metrics.aggregates"
- exchange_type = "direct"
- durable = True
+ exchange = Exchange("vumi.metrics.aggregates",
+ exchange_type="direct", durable=True)
  routing_key = "vumi.metrics.aggregates"
 
  def __init__(self, callback):
@@ -45,9 +44,8 @@ def consume_message(self, vumi_message):
 class AggregatedMetricPublisher(Publisher):
  """Publishes aggregated metrics.
  """
- exchange_name = "vumi.metrics.aggregates"
- exchange_type = "direct"
- durable = True
+ exchange = Exchange("vumi.metrics.aggregates",
+ exchange_type="direct", durable=True)
  routing_key = "vumi.metrics.aggregates"
 
  def publish_aggregate(self, metric_name, timestamp, value):
@@ -70,9 +68,8 @@ class TimeBucketConsumer(Consumer):
  aggregator (list of aggregator names) and values (a
  list of timestamp and value pairs).
  """
- exchange_name = "vumi.metrics.buckets"
- exchange_type = "direct"
- durable = True
+ exchange = Exchange("vumi.metrics.buckets",
+ exchange_type="direct", durable=True)
  ROUTING_KEY_TEMPLATE = "bucket.%d"
 
  def __init__(self, bucket, callback):
@@ -97,9 +94,8 @@ class TimeBucketPublisher(Publisher):
  bucket_size : int, in seconds
  Size of each time bucket in seconds.
  """
- exchange_name = "vumi.metrics.buckets"
- exchange_type = "direct"
- durable = True
+ exchange = Exchange("vumi.metrics.buckets",
+ exchange_type="direct", durable=True)
  ROUTING_KEY_TEMPLATE = "bucket.%d"
 
  def __init__(self, buckets, bucket_size):
@@ -281,9 +277,7 @@ def consume_metrics(self, metric_name, values):
 class GraphitePublisher(Publisher):
  """Publisher for sending messages to Graphite."""
 
- exchange_name = "graphite"
- exchange_type = "topic"
- durable = True
+ exchange = Exchange("graphite", exchange_type="topic", durable=True)
  auto_delete = False
  delivery_mode = 2
  require_bind = False # Graphite uses a topic exchange