2.14.0 rewrite

3 files changed, 30 insertions, 33 deletions

diff --git a/Dockerfile b/Dockerfile
index 3b3974c..3bdedfb 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -4,6 +4,6 @@ RUN mkdir -p /usr/src/app
 WORKDIR /usr/src/app
 
 COPY . /usr/src/app
-RUN apk add --no-cache git && pip install --no-cache-dir -r requirements.txt && pip install --no-cache-dir git+git://github.com/zzzeek/sqlalchemy
+RUN pip install --pre --no-cache-dir -r requirements.txt
 
 CMD ["python", "worker.py"]
diff --git a/requirements.txt b/requirements.txt
index f73baba..0134b05 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,6 @@
-appdirs==1.4.3
-cymysql==0.8.9
-gmpy==1.17
+cymysql==0.9.0
 mpmath==0.19
-packaging==16.8
 pika==0.10.0
+six==1.10.0
+SQLAlchemy==1.2.0b1
 trueskill==0.4.4
diff --git a/worker.py b/worker.py
index 0d8f3c4..c1d2704 100644
--- a/worker.py
+++ b/worker.py
@@ -47,9 +47,6 @@ db = rabbit = channel = None
 queue = []
 timer = None
 
-# models
-mvpmodel = None
-
 
 def connect():
     global Match, Roster, Participant, ParticipantStats, Player
@@ -58,13 +55,7 @@ def connect():
     # generate schema from db
     Base = automap_base()
     engine = create_engine(DATABASE_URI, pool_size=1, pool_recycle=3600)
-    while True:
-        try:
-            Base.prepare(engine, reflect=True)
-            break
-        except OperationalError as err:
-            logging.error(err)
-            time.sleep(5)
+    Base.prepare(engine, reflect=True)
 
     # definitions
     # TODO check whether the primaryjoin clause is the best method to do this
@@ -100,15 +91,10 @@ def connect():
 
     db = Session(engine)
 
-    while True:
-        try:
-            rabbit = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URI))
-            break
-        except pika.exceptions.ConnectionClosed as err:
-            logging.error(err)
-            time.sleep(5)
+    rabbit = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URI))
     channel = rabbit.channel()
     channel.queue_declare(queue=QUEUE, durable=True)
+    channel.queue_declare(queue=QUEUE + "_failed", durable=True)
     channel.basic_qos(prefetch_count=BATCHSIZE)
     channel.basic_consume(newjob, queue=QUEUE)
 
@@ -117,21 +103,36 @@ def newjob(_, method, properties, body):
     global timer, queue
     queue.append((method, properties, body))
     if timer is None:
-        timer = rabbit.add_timeout(IDLE_TIMEOUT, process)
+        timer = rabbit.add_timeout(IDLE_TIMEOUT, try_process)
     if len(queue) == BATCHSIZE:
-        process()
-
+        try_process()
 
-def process():
-    global timer, queue, db, mvpmodel
+def try_process():
+    global timer, queue
     if timer is not None:
         rabbit.remove_timeout(timer)
         timer = None
-    jobs = queue[:]
+    try:
+        process()
+    except e:
+        logging.error(e)
+        for q in queue:
+            # move to error queue and NACK
+            channel.basic_publish("", QUEUE + "_failed", q[1], q[2])
+            channel.basic_nack(q[0].delivery_tag, requeue=False)
+        queue = []
+        return
+
+    logging.info("acking batch")
+    for q in queue:
+        channel.basic_ack(q[0].delivery_tag)
     queue = []
 
-    logging.info("analyzing batch %s", str(len(jobs)))
-    ids = list(set([str(id, "utf-8") for _, _, id in jobs]))
+
+def process():
+    global timer, queue, db
+    logging.info("analyzing batch %s", str(len(queue)))
+    ids = list(set([str(id, "utf-8") for _, _, id in queue]))
 
     with db.no_autoflush:
         env = trueskill.TrueSkill(
@@ -206,9 +207,6 @@ def process():
                         player.trueskill_sigma = rating.sigma
 
     db.commit()
-    # ack all until this one
-    logging.info("acking batch")
-    channel.basic_ack(jobs[-1][0].delivery_tag, multiple=True)
     # notify web
     for api_id in ids:
         channel.basic_publish("amq.topic", "participant." + api_id,