Permalink
Browse files

Adding emdr-gateway and emdr-snooper bin convenience commands.

  • Loading branch information...
1 parent a7354ee commit c8f0e2de3ec6453bde21dc92341097ef94f16cd8 Greg Taylor committed Apr 16, 2012
Showing with 86 additions and 21 deletions.
  1. +32 −0 bin/emdr-gateway
  2. +39 −0 bin/emdr-snooper
  3. +1 −1 bin/fake_history.py
  4. +1 −1 bin/fake_order.py
  5. +8 −1 emdr/daemons/gateway/order_pusher.py
  6. +3 −18 emdr/daemons/gateway/wsgi.py
  7. +2 −0 setup.py
View
@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+"""
+WSGI gateway application. Runs using bottle+gevent.
+"""
+import argparse
+from bottle import run
+from emdr.conf import default_settings
+
+parser = argparse.ArgumentParser(
+ description="The gateway accepts uploads from player uploader clients " \
+ "over HTTP. Market data is then sent to the broker.",
+)
+parser.add_argument(
+ '--sender', action='append', dest='senders',
+ help="Overrides default gateway sender bindings. This determines where " \
+ "the gateway sends its messages, typically a broker.")
+
+parsed = parser.parse_args()
+
+if parsed.senders:
+ print("* Overriding default gateway sender bindings.")
+ default_settings.GATEWAY_SENDER_BINDINGS = parsed.senders
+
+#noinspection PyUnresolvedReferences
+from emdr.daemons.gateway import wsgi
+
+# Fire up a bottle+gevent process.
+run(
+ host='localhost',
+ port=8080,
+ server='gevent',
+)
View
@@ -0,0 +1,39 @@
+#!/usr/bin/env python
+"""
+CLI util used to attach to ZMQ sockets to listen to whatever is coming down.
+Currently hard-wired to SUB.
+"""
+import argparse
+import zlib
+import zmq
+import simplejson
+from pprint import pprint
+
+parser = argparse.ArgumentParser(
+ description="Connects to a PUB ZMQ socket and prints whatever is coming" \
+ "out. ",
+)
+parser.add_argument('receiver', nargs=1, help="The ZMQ socket to connect to.")
+
+parsed = parser.parse_args()
+
+receiver_uri = parsed.receiver[0]
+
+context = zmq.Context()
+subscriber = context.socket(zmq.SUB)
+
+# Connect to the first publicly available relay.
+subscriber.connect(receiver_uri)
+# Disable filtering.
+subscriber.setsockopt(zmq.SUBSCRIBE, "")
+
+print("Connected to %s" % receiver_uri)
+
+while True:
+ # Receive raw market JSON strings.
+ market_json = zlib.decompress(subscriber.recv())
+ # Un-serialize the JSON data to a Python dict.
+ market_data = simplejson.loads(market_json)
+ # Dump the market data to stdout. Or, you know, do more fun
+ # things here.
+ pprint(market_data)
View
@@ -33,7 +33,7 @@
data = simplejson.dumps(data)
r = requests.post(
- 'http://localhost:8080/emdr/upload/unified/',
+ 'http://localhost:8080/upload/unified/',
data=data,
)
print "RESPONSE"
View
@@ -44,7 +44,7 @@
data = simplejson.dumps(data)
r = requests.post(
- 'http://localhost:8080/emdr/upload/unified/',
+ 'http://localhost:8080/upload/unified/',
data=data,
)
print "RESPONSE"
@@ -6,6 +6,7 @@
import logging
import zlib
import simplejson
+import gevent
from gevent.queue import Queue
from gevent_zeromq import zmq
from emdr.conf import default_settings as settings
@@ -21,7 +22,9 @@
# Get the list of transports to bind from settings. This allows us to listen
# for processor connections from multiple places (UNIX sockets + TCP sockets).
# By default, we only listen for UNIX domain sockets.
+logger.info("Order data will be sent to:")
for binding in settings.GATEWAY_SENDER_BINDINGS:
+ logger.info(" * %s" % binding)
sender.connect(binding)
def worker():
@@ -46,4 +49,8 @@ def worker():
sender.send(compressed_msg)
logger.info('Pushed message of length %s' % len(compressed_msg))
-
+# Fire up gevent workers that send raw market order data to processor processes
+# in the background without blocking the WSGI app.
+logger.info("Spawning %d PUSH workers." % settings.NUM_GATEWAY_SENDER_WORKERS)
+for worker_num in range(settings.NUM_GATEWAY_SENDER_WORKERS):
+ gevent.spawn(worker)
@@ -12,7 +12,7 @@
from logging.config import dictConfig
from emdr.conf import default_settings as settings
dictConfig(settings.LOGGING)
-logger = logging.getLogger('src.daemons.gateway.wsgi')
+logger = logging.getLogger('emdr.daemons.gateway.wsgi')
import gevent
from gevent import monkey; gevent.monkey.patch_all()
@@ -53,6 +53,7 @@ def upload_eve_marketeer():
# The job dict gets shoved into a gevent queue, where it awaits sending
# to the processors via the src.daemons.gateway.order_pusher module.
order_pusher.order_upload_queue.put(job_dict)
+ logger.info("Accepted upload from %s" % request.remote_addr)
# Goofy, but apparently expected by EVE Market Data Uploader.
return '1'
@@ -73,23 +74,7 @@ def upload_eve_marketeer():
# The job dict gets shoved into a gevent queue, where it awaits sending
# to the processors via the src.daemons.gateway.order_pusher module.
order_pusher.order_upload_queue.put(job_dict)
+ logger.info("Accepted upload from %s" % request.remote_addr)
# Goofy, but apparently expected by EVE Market Data Uploader.
return '1'
-
-# Fire up gevent workers that send raw market order data to processor processes
-# in the background without blocking the WSGI app.
-for worker_num in range(settings.NUM_GATEWAY_SENDER_WORKERS):
- logger.info("Spawning Gateway->Processor PUSH worker.")
- gevent.spawn(order_pusher.worker)
-
-if __name__ == '__main__':
- # Start the built-in Bottle server for development, for now.
- run(
- host='localhost',
- port=8080,
- server='gevent',
- )
-else:
- # gunicorn will eventually use this in production.
- application = default_app()
View
@@ -33,8 +33,10 @@
scripts = [
'bin/emdr-announcer',
'bin/emdr-broker',
+ 'bin/emdr-gateway',
'bin/emdr-processor',
'bin/emdr-relay',
+ 'bin/emdr-snooper',
'bin/ec-feeder',
]

0 comments on commit c8f0e2d

Please sign in to comment.