Django Celery create task with JS microservice

Question

Hello I have a Django app running celery, and I'm trying to add tasks to the queue using a microservice in JS. When I add the task to AWS SQS though Django everything works fine, but when I use JS i get this error:

[2021-01-05 19:39:55,982: WARNING/MainProcess] Received and deleted unknown message.  Wrong destination?!?

The full contents of the message body was: body: '{"expires":null,"utc":true,"args":[5456,2878],"chord":null,"callbacks":null,"errbacks":null,"taskset":null,"id":"1a361c85-2209-4ffa-95c2-ee2e4855155e","retries":0,"task":"config.celery.debug_task","timelimit":[null,null],"eta":null,"kwargs":{}}' (244b)
{content_type:None content_encoding:None
  delivery_info:{'sqs_message': {'MessageId': '7b7d2948-c069-4f8b-9fdc-9c068d52f463', 'ReceiptHandle': 'AQEBxhqW2sRWf+Z851fw7nqRX6MQFVcTfjH5xqiIgYIiMa3AN3R235VxhM8pM7mcByw3eOZ3Y7kH5oZ+noFVzfjSllgnoh8idB/V7WWY2urNHKJrQadRT5cf4NcUVkFmB8+d2rLiAXuuyqpGbEMvmx1Dn49/5C3Fx8Eq+eUyB1oeilIrCqfMvIkG/yX5TdedxM9B2VBThZ/XtHqrgYCkJvEt9ozssM0f+INRHUrpVQMYCmUX9aTWeWljrTOapMTg27M6aie6HaDQxLK0FJvZUNr2d0uJhZ4C2qRGWrSo2VpD7QK7pslltZ12PVHKPw9X+cBGdWwJrdh5I0fBITuoy+CUUnybDekz668jJnsf1gcmpx8cBoVrMLocPi753g2klGf++mbFeL7yjENzb1YqZrrfvg==', 'MD5OfBody': '9bb39da667d1e840f8532a74a8dcecaa', 'Body': 'eyJleHBpcmVzIjpudWxsLCJ1dGMiOnRydWUsImFyZ3MiOls1NDU2LDI4NzhdLCJjaG9yZCI6bnVsbCwiY2FsbGJhY2tzIjpudWxsLCJlcnJiYWNrcyI6bnVsbCwidGFza3NldCI6bnVsbCwiaWQiOiIxYTM2MWM4NS0yMjA5LTRmZmEtOTVjMi1lZTJlNDg1NTE1NWUiLCJyZXRyaWVzIjowLCJ0YXNrIjoiY29uZmlnLmNlbGVyeS5kZWJ1Z190YXNrIiwidGltZWxpbWl0IjpbbnVsbCxudWxsXSwiZXRhIjpudWxsLCJrd2FyZ3MiOnt9fQ=='}, 'sqs_queue': 'SQS_QUEUE_URL_HERE'} headers={}}

I am using this code to send the message:

let taskId = uuidv4();
let result = {
    "expires": null,
    "utc": true,
    "args": [5456, 2878],
    "chord": null,
    "callbacks": null,
    "errbacks": null,
    "taskset": null,
    "id": taskId,
    "retries": 0,
    "task": "config.celery.debug_task",
    "timelimit": [null, null],
    "eta": null,
    "kwargs": {}
}

const client = new SQSClient({
    region: "eu-west-3",
    credentialDefaultProvider: myCredentialProvider
});

const send = new SendMessageCommand({
    // use wrangler secrets to provide this global variable
    QueueUrl: "SQS_QUEUE_URL_HERE",
    MessageBody: Buffer.from(JSON.stringify(result)).toString("base64")
});

let resultSQS = client.send(send);

I debugged the django task payload to copy it, so I'm sending the same data it needs, but getting this error. Anyone knows if I'm missing something?

Thanks

Answer 1

I solved the problem, the MessageBody was wrong. This is the solution

let taskId = uuidv4();
    let body = [[{"hello": "world from JS", "success": true}], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]

    let properties = {
        "correlation_id": taskId,
        "reply_to": "5480744c-f3ca-3e3f-9403-445fa5b865e1",
        "delivery_mode": 2,
        "delivery_info": {
            "exchange": "",
            "routing_key": "default"
        },
        "priority": 0,
        "body_encoding": "base64",
        "delivery_tag": uuidv4()
    }


    let headers = {
        "lang": "py",
        "task": "config.celery.debug_task",
        "id": taskId,
        "shadow": null,
        "eta": null,
        "expires": null,
        "group": null,
        "retries": 0,
        "timelimit": [null, null],
        "root_id": taskId,
        "parent_id": null,
        "argsrepr": "({'hello': 'wolrd', 'success': True},)",
        "kwargsrepr": "{}",
        "origin": "gen21089@Marcoss-MacBook-Pro.local"
    }


    let payload = {
        "body": Buffer.from(JSON.stringify(body)).toString("base64"),
        "content-encoding": "utf-8",
        "content-type": "application/json",
        "headers": headers,
        "properties": properties
    }


    let encodedPayload = Buffer.from(JSON.stringify(payload)).toString("base64")

    const client = new SQSClient({
        region: "eu-west-3",
        credentialDefaultProvider: myCredentialProvider
    });

    const sendData = new SendMessageCommand({
        // use wrangler secrets to provide this global variable
        QueueUrl: "URL",
        MessageBody: encodedPayload
    });

    let data = await client.send(sendData);

So the payload in base64 contains a "body" that is also in base64 and headers that are what Celery reads to locate the task.