Merge 544dcc6 into 8a8db0f

config/queue.py

@@ -26,4 +26,12 @@
 |
 """
 
-DRIVERS = {}
+DRIVERS = {
+    'amqp': {
+        'username': 'guest',
+        'password': 'guest',
+        'host': 'localhost',
+        'port': '5672',
+        'channel': 'default',
+    }
+}

masonite/commands/QueueWorkCommand.py

@@ -14,9 +14,15 @@ def callback(ch, method, properties, body):
     job = pickle.loads(body)
     obj = job['obj']
     args = job['args']
+    callback = job['callback']
     if inspect.isclass(obj):
         obj = container.resolve(obj)
-    obj.handle(*args)
+
+    try:
+        getattr(obj, callback)(*args)
+    except AttributeError:
+        obj(*args)
+
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
 

masonite/drivers/QueueAmqpDriver.py

@@ -60,7 +60,7 @@ def _publish(self, body):
                                        delivery_mode=2,  # make message persistent
                                    ))
 
-    def push(self, *objects, args=()):
+    def push(self, *objects, args=(), callback='handle'):
         """Push objects onto the amqp stack.
 
         Arguments:
@@ -70,7 +70,7 @@ def push(self, *objects, args=()):
         for obj in objects:
             # Publish to the channel for each object
             try:
-                self._publish({'obj': obj, 'args': args})
+                self._publish({'obj': obj, 'args': args, 'callback': callback})
             except self.pika.exceptions.ConnectionClosed:
                 self._connect()
                 self._publish({'obj': obj, 'args': args})

masonite/drivers/QueueAsyncDriver.py

@@ -19,7 +19,7 @@ def __init__(self, app: App):
         """
         self.container = app
 
-    def push(self, *objects, args=()):
+    def push(self, *objects, args=(), callback='handle'):
         """Push objects onto the async stack.
 
         Arguments:
@@ -29,6 +29,12 @@ def push(self, *objects, args=()):
             if inspect.isclass(obj):
                 obj = self.container.resolve(obj)
 
-            thread = threading.Thread(
-                target=obj.handle, args=args, kwargs={})
+            try:
+                thread = threading.Thread(
+                    target=getattr(obj, callback), args=args, kwargs={})
+            except AttributeError:
+                # Could be wanting to call only a method asyncronously
+                thread = threading.Thread(
+                    target=obj, args=args, kwargs={})
+
             thread.start()

masonite/info.py

@@ -1,3 +1,3 @@
 """Module for specifying the Masonite version in a central location."""
 
-VERSION = '2.1.0b3.post7'
+VERSION = '2.1.0'

masonite/queues/ShouldQueue.py

@@ -0,0 +1,2 @@
+class ShouldQueue:
+    pass

tests/queues/test_drivers.py

@@ -0,0 +1,55 @@
+from masonite.app import App
+from masonite.drivers import QueueAsyncDriver, QueueAmqpDriver
+from masonite.managers import QueueManager
+from config import queue
+
+from masonite.queues.Queueable import Queueable
+import os
+from masonite.environment import LoadEnvironment, env
+
+LoadEnvironment()
+
+
+class Job(Queueable):
+
+    def handle(self):
+        print('sending from job handled')
+        return 'test'
+
+class Random(Queueable):
+
+    def send(self):
+        print('sending from random send method')
+        return 'test'
+
+    def handle(self):
+        print('sending from random handle method')
+        return 'test'
+
+
+class TestAsyncDriver:
+
+    def setup_method(self):
+        self.app = App()
+
+        self.app.bind('QueueAsyncDriver', QueueAsyncDriver)
+        self.app.bind('QueueAmqpDriver', QueueAmqpDriver)
+        self.app.bind('QueueConfig', queue)
+        self.app.bind('Queueable', Queueable)
+        self.app.bind('Container', self.app)
+        self.app.bind('QueueManager', QueueManager(self.app))
+        self.drivers = ['async']
+        if env('RUN_AMQP'):
+            self.drivers.append('amqp')
+
+    def test_async_driver_pushes_to_queue(self):
+        for driver in self.drivers:
+            assert self.app.make('QueueManager').driver(driver).push(Job) is None
+
+    def test_async_driver_can_run_any_callback_method(self):
+        for driver in self.drivers:
+            assert self.app.make('QueueManager').driver(driver).push(Random, callback="send") is None
+
+    def test_async_driver_can_run_any_method(self):
+        for driver in self.drivers:
+            assert self.app.make('QueueManager').driver(driver).push(Random().send) is None