run-llama/llama_deploy
1
0
Fork 0
mirror of https://github.com/run-llama/llama_deploy.git synced 2026-08-24 21:01:26 -04:00
Code Issues 37 Packages Projects Releases 10 Wiki Activity

[PR #511] [MERGED] feat: refactor message queue communication strategy #512

New Issue
Closed
opened 2026-02-16 02:15:25 -05:00 by yindo · 0 comments
yindo commented 2026-02-16 02:15:25 -05:00
Owner
Copy Link
Copy Source

📋 Pull Request Information

Original PR: https://github.com/run-llama/llama_deploy/pull/511
Author: @masci
Created: 5/25/2025
Status: ✅ Merged
Merged: 5/27/2025
Merged by: @masci

Base: main ← Head: massi/message-queue-refact

📝 Commits (10+)

📊 Changes

38 files changed (+1053 additions, -1212 deletions)

View changed files

📝 e2e_tests/apiserver/conftest.py (+0 -1)
📝 e2e_tests/apiserver/test_hitl.py (+7 -10)
📝 e2e_tests/message_queues/aws/conftest.py (+17 -11)
📝 e2e_tests/message_queues/aws/test_message_queue.py (+6 -24)
➖ e2e_tests/message_queues/conftest.py (+0 -20)
📝 e2e_tests/message_queues/kafka/test_message_queue.py (+6 -23)
📝 e2e_tests/message_queues/rabbitmq/test_message_queue.py (+7 -19)
📝 e2e_tests/message_queues/redis/test_message_queue.py (+9 -22)
📝 e2e_tests/message_queues/simple/test_message_queue.py (+9 -24)
📝 llama_deploy/apiserver/deployment.py (+4 -20)
📝 llama_deploy/client/models/apiserver.py (+4 -2)
📝 llama_deploy/control_plane/server.py (+22 -10)
📝 llama_deploy/deploy/deploy.py (+1 -21)
➖ llama_deploy/message_consumers/__init__.py (+0 -3)
➖ llama_deploy/message_consumers/remote.py (+0 -37)
📝 llama_deploy/message_queues/apache_kafka.py (+17 -43)
📝 llama_deploy/message_queues/aws.py (+60 -84)
📝 llama_deploy/message_queues/base.py (+11 -26)
📝 llama_deploy/message_queues/rabbitmq.py (+47 -66)
📝 llama_deploy/message_queues/redis.py (+30 -87)

...and 18 more files

📄 Description

This PR rolls out a major refactoring of the way the Control Plane and the Workflow Service talk to the message queue.

This was the sequence of the operations before:

  • ControlPlane calls a method on the MessageQueue to register itself as a consumer
  • MessageQueue returns a callable that makes a POST request to the ControlPlane api at every message received
  • ControlPlane must run that callable as a background task
  • ControlPlane must run dedicated HTTP endpoints for the callable

This is the sequence of the operations after this PR:

  • ControlPlane calls a method on the MessageQueue to register itself as a consumer
  • MessageQueue returns a client object with a generic API but concrete implementation for the specific message queue type
  • ControlPlane uses the client to consume messages

Advantages of this approach:

  • No more HTTP calls involved in exchanging messages, much faster
  • WorkflowService doesn't need any HTTP-server functionality anymore, required code is a fraction now
  • Consumers can register as many times as they want, the MessageQueue doesn't care anymore
  • Code for MessageQueue is much simpler now

🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.

## 📋 Pull Request Information **Original PR:** https://github.com/run-llama/llama_deploy/pull/511 **Author:** [@masci](https://github.com/masci) **Created:** 5/25/2025 **Status:** ✅ Merged **Merged:** 5/27/2025 **Merged by:** [@masci](https://github.com/masci) **Base:** `main` ← **Head:** `massi/message-queue-refact` --- ### 📝 Commits (10+) - [`c8febca`](https://github.com/run-llama/llama_deploy/commit/c8febcace7824822435c6e15315adf7f7843d016) feat: refactor message queue communication - [`a0a0295`](https://github.com/run-llama/llama_deploy/commit/a0a0295164dc2d04cb00f1238bb65c2a3277e1c5) remove unused code - [`618bc14`](https://github.com/run-llama/llama_deploy/commit/618bc141126cd05b2d99514c7be83f8cb32ed3bb) fix unit tests - [`10cb270`](https://github.com/run-llama/llama_deploy/commit/10cb270eff78f84209715692c2aa43089ad7bc36) clean shutdown on message consuming - [`998c902`](https://github.com/run-llama/llama_deploy/commit/998c90286f0cce58b1351ee41638d18e61ccfbc6) fix roundtrip tests - [`411c7ab`](https://github.com/run-llama/llama_deploy/commit/411c7ab2cd551576a127ef18a1e7ebfee7a81365) fix more e2e - [`9f8933f`](https://github.com/run-llama/llama_deploy/commit/9f8933fcc10da2dfb10ba6664babe2c52bcd1c52) fix reload operation - [`acdf2a1`](https://github.com/run-llama/llama_deploy/commit/acdf2a1d376defffd1fceb46200e1f4c7d93b3f2) fix rabbit roundtrip - [`82d22f3`](https://github.com/run-llama/llama_deploy/commit/82d22f3296ffa809a1f29a622dd4ccead65bfe3e) remove message consumer - [`27bbf1f`](https://github.com/run-llama/llama_deploy/commit/27bbf1f2d8ca788f45797ec3337ad87fe59a7aae) fix e2e tests ### 📊 Changes **38 files changed** (+1053 additions, -1212 deletions) <details> <summary>View changed files</summary> 📝 `e2e_tests/apiserver/conftest.py` (+0 -1) 📝 `e2e_tests/apiserver/test_hitl.py` (+7 -10) 📝 `e2e_tests/message_queues/aws/conftest.py` (+17 -11) 📝 `e2e_tests/message_queues/aws/test_message_queue.py` (+6 -24) ➖ `e2e_tests/message_queues/conftest.py` (+0 -20) 📝 `e2e_tests/message_queues/kafka/test_message_queue.py` (+6 -23) 📝 `e2e_tests/message_queues/rabbitmq/test_message_queue.py` (+7 -19) 📝 `e2e_tests/message_queues/redis/test_message_queue.py` (+9 -22) 📝 `e2e_tests/message_queues/simple/test_message_queue.py` (+9 -24) 📝 `llama_deploy/apiserver/deployment.py` (+4 -20) 📝 `llama_deploy/client/models/apiserver.py` (+4 -2) 📝 `llama_deploy/control_plane/server.py` (+22 -10) 📝 `llama_deploy/deploy/deploy.py` (+1 -21) ➖ `llama_deploy/message_consumers/__init__.py` (+0 -3) ➖ `llama_deploy/message_consumers/remote.py` (+0 -37) 📝 `llama_deploy/message_queues/apache_kafka.py` (+17 -43) 📝 `llama_deploy/message_queues/aws.py` (+60 -84) 📝 `llama_deploy/message_queues/base.py` (+11 -26) 📝 `llama_deploy/message_queues/rabbitmq.py` (+47 -66) 📝 `llama_deploy/message_queues/redis.py` (+30 -87) _...and 18 more files_ </details> ### 📄 Description This PR rolls out a major refactoring of the way the Control Plane and the Workflow Service talk to the message queue. This was the sequence of the operations before: - ControlPlane calls a method on the MessageQueue to register itself as a consumer - MessageQueue returns a callable that makes a POST request to the ControlPlane api at every message received - ControlPlane must run that callable as a background task - ControlPlane must run dedicated HTTP endpoints for the callable This is the sequence of the operations after this PR: - ControlPlane calls a method on the MessageQueue to register itself as a consumer - MessageQueue returns a client object with a generic API but concrete implementation for the specific message queue type - ControlPlane uses the client to consume messages Advantages of this approach: - No more HTTP calls involved in exchanging messages, much faster - WorkflowService doesn't need any HTTP-server functionality anymore, required code is a fraction now - Consumers can register as many times as they want, the MessageQueue doesn't care anymore - Code for MessageQueue is much simpler now --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>
yindo added the pull-request label 2026-02-16 02:15:25 -05:00
yindo closed this issue 2026-02-16 02:15:25 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
P0
P1
P2
bug
documentation
documentation
enhancement
example
good first issue
invalid
pull-request
Mirrored from GitHub Pull Request
 question
topic:CLI
topic:SDK
topic:SDK
topic:SDK
topic:apiserver
topic:control_plane
topic:message_queues
wontfix
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: run-llama/llama_deploy#512
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?