New feature: message with priority

celery-java/src/main/java/com/geneea/celery/Celery.java

@@ -55,15 +55,23 @@ public class Celery {
     private Celery(final String brokerUri,
                    @Nullable final String queue,
                    @Nullable final String backendUri,
-                   @Nullable final ExecutorService executor) {
+                   @Nullable final ExecutorService executor,
+                   @Nullable final boolean isPriQueue,
+                   @Nullable final int maxPriority) {
         this.queue = queue == null ? "celery" : queue;
 
         ExecutorService executorService = executor != null ? executor : Executors.newCachedThreadPool();
 
         broker = Suppliers.memoize(() -> {
             Broker b = CeleryBrokers.createBroker(brokerUri, executorService);
             try {
-                b.declareQueue(Celery.this.queue);
+                if(isPriQueue  && maxPriority != 0){
+                    b.declarePriQueue(Celery.this.queue, maxPriority);
+                }
+                else {
+                    b.declareQueue(Celery.this.queue);
+                }
+
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
@@ -110,6 +118,22 @@ public AsyncResult<?> submit(Class<?> taskClass, String method, Object[] args) t
         return submit(taskClass.getName() + "#" + method, args);
     }
 
+    /**
+     * Submit a Java task for processing with priority. You'll probably not need to call this method. rather use @{@link CeleryTask}
+     * annotation.
+     *
+     * @param taskClass task implementing class
+     * @param method method in {@code taskClass} that does the work
+     * @param priority the priority of the task
+     * @param args positional arguments for the method (need to be JSON serializable)
+     * @return asynchronous result
+     *
+     * @throws IOException if the message couldn't be sent
+     */
+    public AsyncResult<?> submitWithPri(Class<?> taskClass, String method, int priority, Object[] args) throws IOException {
+        return submitWithPri(taskClass.getName() + "#" + method, priority, args);
+    }
+
     /**
      * Submit a task by name. A low level method for submitting arbitrary tasks that don't have their proxies
      * generated by @{@link CeleryTask} annotation.
@@ -166,6 +190,61 @@ public AsyncResult<?> submit(String name, Object[] args) throws IOException {
         return new AsyncResultImpl<>(result);
     }
 
+    /**
+     * Submit a task by name with priority.
+     *
+     * @param name task name as understood by the worker
+     * @param priority the priority of the message
+     * @param args positional arguments for the method (need to be JSON serializable)
+     * @return asynchronous result
+     * @throws IOException
+     */
+    public AsyncResult<?> submitWithPri(String name, int priority, Object[] args) throws IOException {
+        // Get the provider early to increase the chance to find out there is a connection problem before actually
+        // sending the message.
+        //
+        // This will help for example in the case when the connection can't be established at all. The connection may
+        // still drop after sending the message but there isn't much we can do about it.
+        Optional<Backend.ResultsProvider> rp = resultsProvider.get();
+        String taskId = UUID.randomUUID().toString();
+
+        ArrayNode payload = jsonMapper.createArrayNode();
+        ArrayNode argsArr = payload.addArray();
+        for (Object arg : args) {
+            argsArr.addPOJO(arg);
+        }
+        payload.addObject();
+        payload.addObject()
+                .putNull("callbacks")
+                .putNull("chain")
+                .putNull("chord")
+                .putNull("errbacks");
+
+        Message message = broker.get().newMessageWithPriority(priority);
+        message.setBody(jsonMapper.writeValueAsBytes(payload));
+        message.setContentEncoding("utf-8");
+        message.setContentType("application/json");
+
+        Message.Headers headers = message.getHeaders();
+        headers.setId(taskId);
+        headers.setTaskName(name);
+        headers.setArgsRepr("(" + Joiner.on(", ").join(args) + ")");
+        headers.setOrigin(clientName);
+        if (rp.isPresent()) {
+            headers.setReplyTo(clientId);
+        }
+
+        message.send(queue);
+
+        Future<Object> result;
+        if (rp.isPresent()) {
+            result = rp.get().getResult(taskId);
+        } else {
+            result = CompletableFuture.completedFuture(null);
+        }
+        return new AsyncResultImpl<>(result);
+    }
+
     public interface AsyncResult<T> {
         boolean isDone();
 

celery-java/src/main/java/com/geneea/celery/brokers/rabbit/RabbitBroker.java

@@ -22,18 +22,41 @@ public void declareQueue(String name) throws IOException {
         channel.queueDeclare(name, true, false, false, null);
     }
 
+    @Override
+    public void declarePriQueue(String name, int maxPriority) throws IOException {
+        Map<String, Object> props = new HashMap<>();
+        props.put("x-max-priority", maxPriority);
+        channel.queueDeclare(name, true, false, false, props);
+    }
+
     @Override
     public Message newMessage() {
         return new RabbitMessage();
     }
 
+    @Override
+    public Message newMessageWithPriority(int priority) {
+        return new RabbitMessage(priority);
+    }
+
     class RabbitMessage implements Message {
         private byte[] body;
-        private final AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder()
-                .deliveryMode(2)
-                .priority(0);
+        private final AMQP.BasicProperties.Builder props;
+
         private final RabbitMessageHeaders headers = new RabbitMessageHeaders();
 
+        public RabbitMessage(){
+            props = new AMQP.BasicProperties.Builder()
+                    .deliveryMode(2)
+                    .priority(0);
+        }
+
+        public RabbitMessage(int priority){
+            props = new AMQP.BasicProperties.Builder()
+                    .deliveryMode(2)
+                    .priority(priority);
+        }
+
         @Override
         public void setBody(byte[] body) {
             this.body = body;

celery-java/src/main/java/com/geneea/celery/spi/Broker.java

@@ -20,8 +20,21 @@ public interface Broker {
      */
     void declareQueue(String name) throws IOException;
 
+    /**
+     * @param name queue name
+     * @param maxPriority the max priority of the queue with priority
+     * @throws IOException
+     */
+    void declarePriQueue(String name, int maxPriority) throws IOException;
+
     /**
      * @return message that can be constructed and later sent
      */
     Message newMessage();
+
+    /**
+     * @param priority the priority of the message that is executed
+     * @return message that can be constructed and later sent
+     */
+    Message newMessageWithPriority(int priority);
 }