broker

#!/usr/bin/python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
import optparse, socket
from brokerlib import Broker
from queue import Queue
from selector import Selector

parser = optparse.OptionParser(usage="usage: %prog [options] QUEUE_1 ... QUEUE_n",
                               description="Prototype amqp broker.")
parser.add_option("-i", "--interface", default="0.0.0.0",
                  help="interface to listen on (default %default)")
parser.add_option("-c", "--container", default=socket.gethostname(),
                  help="container-id for the broker (default %default)")
parser.add_option("-n", "--nodes", default=[], action="append",
                  help="load nodes from specified file")
parser.add_option("-u", "--users", default=[], action="append",
                  help="load user definitions from specified file")
parser.add_option("-p", "--port", type=int, default=5672,
                  help="port to listen on (default %default)")
parser.add_option("-a", "--auth", action="store_true",
                  help="enable sasl authentication layer")
parser.add_option("-f", "--frame-size", type=int, default=4294967295,
                  help="specify the maximum frame size")
parser.add_option("-w", "--window", type=int, default=65536,
                  help="session window size")
parser.add_option("-e", "--period", type=float, default=None,
                  help="update period for session window")
parser.add_option("-t", "--trace", default="err",
                  help="enable tracing for specified categories")
parser.add_option("-T", "--threshold", dest="thresholds", default=[],
                  action="append", metavar="NAME=THRESHOLD",
                  help="specify flow control threshold for a queue")
parser.add_option("-g", "--graphics", action="store_true",
                  help="launch the broker with graphics enabled")

opts, args = parser.parse_args()

if opts.graphics:
  from window import Window
else:
  class Window:
    def __init__(self, *args, **kwargs):
      pass
    def add(self, *args, **kwargs):
      raise RuntimeError("graphics are not enabled")
    def redraw(self):
      pass

try:
  mechanisms = ["ANONYMOUS", "PLAIN"]
  passwords = {}
  for u in opts.users:
    exec open(u) in globals(), passwords

  broker = Broker(opts.container)
  broker.window = opts.window
  broker.period = opts.period
  broker.frame_size = opts.frame_size
  broker.auth = opts.auth
  broker.mechanisms = mechanisms
  broker.passwords = passwords
  broker.traces = opts.trace.split()

  window = Window(lambda *args: selector.stop())
  nodes = {}
  for n in opts.nodes:
    exec open(n) in globals(), nodes

  for name, value in nodes.items():
    if name.startswith("_"):
      continue
    broker.nodes[name] = value

  thresholds = {}
  for value in opts.thresholds:
    name, threshold = value.split("=", 2)
    thresholds[name] = int(threshold)

  for a in args:
    broker.nodes[a] = Queue(thresholds.get(a))

  selector = Selector()
  broker.listener = window.redraw
  broker.bind(opts.interface, opts.port)
  selector.register(broker)

  if opts.graphics:
    selector.run(window.redraw, 0.1)
  else:
    selector.run()
except KeyboardInterrupt:
  pass