{"action":"closed","issue":{"url":"https://api.github.com/repos/RichardKnop/machinery/issues/419","repository_url":"https://api.github.com/repos/RichardKnop/machinery","labels_url":"https://api.github.com/repos/RichardKnop/machinery/issues/419/labels{/name}","comments_url":"https://api.github.com/repos/RichardKnop/machinery/issues/419/comments","events_url":"https://api.github.com/repos/RichardKnop/machinery/issues/419/events","html_url":"https://github.com/RichardKnop/machinery/issues/419","id":447706568,"node_id":"MDU6SXNzdWU0NDc3MDY1Njg=","number":419,"title":"Use signature's routing key as binding key for AMQP broker with direct exchange type","user":{"login":"Martin91","id":2900644,"node_id":"MDQ6VXNlcjI5MDA2NDQ=","avatar_url":"https://avatars0.githubusercontent.com/u/2900644?v=4","gravatar_id":"","url":"https://api.github.com/users/Martin91","html_url":"https://github.com/Martin91","followers_url":"https://api.github.com/users/Martin91/followers","following_url":"https://api.github.com/users/Martin91/following{/other_user}","gists_url":"https://api.github.com/users/Martin91/gists{/gist_id}","starred_url":"https://api.github.com/users/Martin91/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Martin91/subscriptions","organizations_url":"https://api.github.com/users/Martin91/orgs","repos_url":"https://api.github.com/users/Martin91/repos","events_url":"https://api.github.com/users/Martin91/events{/privacy}","received_events_url":"https://api.github.com/users/Martin91/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":3,"created_at":"2019-05-23T14:46:18Z","updated_at":"2019-06-10T02:06:03Z","closed_at":"2019-06-10T02:06:03Z","author_association":"NONE","body":"### Background\r\nIn our project, we are using RabbitMQ as the broker, and we need to group tasks into different queues.\r\n\r\nI am newbie in both Golang and Machinery, and I am not so sure if it is a problem actually.\r\n\r\n### Environment\r\n1. A clean RabbitMQ vhost, no queue, no exchange before, and we are using direct exchange\r\n2. Golang v1.12\r\n3. Machinery v1.6.2\r\n\r\n### Steps To Reproduce\r\n1. Specify individual routing keys in the signatures of different types of tasks, meanwhile, some tasks kept implicit routing key;\r\n2. Run some tests to push tasks;\r\n3. Visited the RabbitMQ Management console http://localhost:15672/#/queues, I found there were some new queues, it was expected;\r\n4. Then I navigated the exchange page http://localhost:15672/#/exchanges/%2F/machinery_exchange,  I found that all queues were binded to the exchange with the same binding key `machinery_task`, which is exactly configured as a binding key to initialize the Machinery server;\r\n5. Fetch messages from the queues, only messages with implicit routing key (default routing key) were there, it means that I missed messages which sent with a custom routing key\r\n\r\n### What To Expect\r\nFor direct AMQP exchange:\r\n\r\n1. It can declare new queues and bind them to the exchange use dynamically specified routing key.\r\n\r\n### My Idea\r\nI have explored the source code and I think it might be possible to do this:\r\n\r\n1. Pass routing key as both queue name and binding key to method `GetOrOpenConnection`, instead of configured `DefaultQueue` and `bindingKey`;\r\n\r\n  the original code(note: the referenced code is based on v1.6.4) is:\r\n \r\n https://github.com/RichardKnop/machinery/blob/3a69c03c4fa63b8a60157979773b2afed80549a9/v1/brokers/amqp/amqp.go#L202-L208\r\n\r\n  and can be changed to:\r\n```go\r\n// isDirectChange extracted function to be reusable\r\nfunc (b *Broker) isDirectExchange() bool {\r\n\treturn b.GetConfig().AMQP != nil && b.GetConfig().AMQP.ExchangeType == \"direct\"\r\n}\r\n\r\nfunc (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error {\r\n\t// ... something else\r\n\tvar queue string\r\n\tvar bindingKey string\r\n\tif b.isDirectExchange() {\r\n\t\tqueue = signature.RoutingKey\r\n\t\tbindingKey = signature.RoutingKey\r\n\t} else {\r\n\t\tqueue = b.GetConfig().DefaultQueue,\r\n\t\tbindingKey = b.GetConfig().AMQP.BindingKey, // queue binding key\r\n\t}\r\n\r\n\tconnection, err := b.GetOrOpenConnection(\r\n\t\tqueue,\r\n\t\tbindingKey,\r\n\t\tnil,                           // exchange declare args\r\n\t\tnil,                           // queue declare args\r\n\t\tamqp.Table(b.GetConfig().AMQP.QueueBindingArgs), // queue binding args\r\n\t)\r\n\t// something else\r\n}\r\n```"}}
{"actor":{"display_login":"Martin91"}}