[WIP][feature] New proxy to use the multi bundles feature (part1) #716
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add a new proxy to use multi bundles feature. - support exchange declare - support queue declare - support queue bind - support queue unbind - support basic publish - support basic consume
GhostBoyBoy
reviewed
Nov 16, 2022
| bindingsJson = JSON_MAPPER.writeValueAsString(this.bindings); | ||
| } catch (JsonProcessingException e) { | ||
| log.error("Failed to bind queue {} to exchange {}", queue, exchangeName, e); | ||
| return CompletableFuture.failedFuture(e); |
There was a problem hiding this comment.
This failedFuture method was introduced in jdk9, we should support jdk8.
| bindingsJson = JSON_MAPPER.writeValueAsString(this.bindings); | ||
| } catch (JsonProcessingException e) { | ||
| log.error("Failed to unbind queue {} to exchange {}", queue, exchangeName, e); | ||
| return CompletableFuture.failedFuture(e); |
| try { | ||
| client = pulsar.getClient(); | ||
| } catch (PulsarServerException e) { | ||
| return CompletableFuture.failedFuture(e); |
GhostBoyBoy
reviewed
Nov 16, 2022
| data = new byte[message.getDataBuffer().readableBytes()]; | ||
| message.getDataBuffer().readBytes(data); | ||
| for (Destination des : destinations) { | ||
| futures.add(sendMessage(des, data, props)); |
GhostBoyBoy
reviewed
Nov 16, 2022
| public void deleteFailed(ManagedLedgerException exception, Object ctx) { | ||
| log.error("{} Failed to delete message at {}", exchange.getName(), ctx, exception); | ||
| } | ||
| }, entry.getPosition()); |
There was a problem hiding this comment.
The entry has been released, and the 190 line position variable should be used.
There was a problem hiding this comment.
nice cache, fixed
GhostBoyBoy
reviewed
Nov 16, 2022
| log.error("Deserialize entry dataBuffer failed. exchangeName: {}, skip it first.", | ||
| exchange.getName(), e); | ||
| PENDING_SIZE_UPDATER.decrementAndGet(this); | ||
| continue; |
There was a problem hiding this comment.
The entry is not released.
GhostBoyBoy
reviewed
Nov 17, 2022
| } | ||
|
|
||
| @Override | ||
| public void receiveExchangeDeleteOk() { |
There was a problem hiding this comment.
The client needs to be notified of the result.
There was a problem hiding this comment.
This method will be supported later.
|
|
||
| @Override | ||
| public void receiveQueueDeleteOk(long messageCount) { | ||
| // nothing to do |
There was a problem hiding this comment.
The client needs to be notified of the result.
There was a problem hiding this comment.
This method will be supported later.
GhostBoyBoy
reviewed
Nov 17, 2022
| }) | ||
| .thenAccept(position -> { | ||
| if (log.isDebugEnabled()) { | ||
| log.debug("Publish message success, position {}", position.toString()); |
There was a problem hiding this comment.
open Publish confirm, BasicAckBody not handled.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
Currently, the AoP can't leverage the multi bundles feature, exchanges and queues belonging to one AMQP virtual host are owned by one namespace bundle, they must work on one broker. So the broker can't scale up simply to support increasing exchanges and queues, if users want to use all brokers of a cluster, they need to use different virtual hosts for different exchanges and queues, it's very hard to adjust existing applications and it's not friendly to use.
Modifications
Verifying this change
Add new test.
Does this pull request potentially affect one of the following parts:
If
yeswas chosen, please highlight the changesDocumentation
Check the box below.
Need to update docs?
doc-required(If you need help on updating docs, create a doc issue)
no-need-doc(Please explain why)
doc(If this PR contains doc changes)