Skip to content

remove try-with #31203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chenrujun merged 1 commit into from
Sep 29, 2022
Merged

remove try-with #31203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 37 additions & 37 deletions ...-eventhubs/src/main/java/com/azure/spring/messaging/eventhubs/core/EventHubsTemplate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,48 +98,48 @@ public <T> Mono<Void> sendAsync(String destination, Message<T> message) {
}

private Mono<Void> doSend(String destination, List<EventData> events, PartitionSupplier partitionSupplier) {
try (EventHubProducerAsyncClient producer = producerFactory.createProducer(destination)) {
CreateBatchOptions options = buildCreateBatchOptions(partitionSupplier);
AtomicReference<EventDataBatch> currentBatch = new AtomicReference<>(
producer.createBatch(options).block());
Flux.fromIterable(events).flatMap(event -> {
final EventDataBatch batch = currentBatch.get();
try {
if (batch.tryAdd(event)) {
return Mono.empty();
} else {
LOGGER.warn("EventDataBatch is full in the collect process or the first event is "
+ "too large to fit in an empty batch! Max size: {}", batch.getMaxSizeInBytes());
}
} catch (AmqpException e) {
LOGGER.error("Event is larger than maximum allowed size.", e);
EventHubProducerAsyncClient producer = producerFactory.createProducer(destination);
CreateBatchOptions options = buildCreateBatchOptions(partitionSupplier);
AtomicReference<EventDataBatch> currentBatch = new AtomicReference<>(
producer.createBatch(options).block());
Flux.fromIterable(events).flatMap(event -> {
final EventDataBatch batch = currentBatch.get();
try {
if (batch.tryAdd(event)) {
return Mono.empty();
} else {
LOGGER.warn("EventDataBatch is full in the collect process or the first event is "
+ "too large to fit in an empty batch! Max size: {}", batch.getMaxSizeInBytes());
}

return Mono.when(
producer.send(batch),
producer.createBatch(options).map(newBatch -> {
currentBatch.set(newBatch);
// Add the event that did not fit in the previous batch.
try {
if (!newBatch.tryAdd(event)) {
LOGGER.error("Event was too large to fit in an empty batch. Max size:{} ",
newBatch.getMaxSizeInBytes());
}
} catch (AmqpException e) {
LOGGER.error("Event was too large to fit in an empty batch. Max size:{}",
newBatch.getMaxSizeInBytes(), e);
} catch (AmqpException e) {
LOGGER.error("Event is larger than maximum allowed size.", e);
return Mono.empty();
}

return Mono.when(
producer.send(batch),
producer.createBatch(options).map(newBatch -> {
currentBatch.set(newBatch);
// Add the event that did not fit in the previous batch.
try {
if (!newBatch.tryAdd(event)) {
LOGGER.error("Event was too large to fit in an empty batch. Max size:{} ",
newBatch.getMaxSizeInBytes());
}
} catch (AmqpException e) {
LOGGER.error("Event was too large to fit in an empty batch. Max size:{}",
newBatch.getMaxSizeInBytes(), e);
}

return newBatch;
}));
})
.then()
.block();
return newBatch;
}));
})
.then()
.block();

@chenrujun chenrujun Sep 29, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not async, please fix it in another PR.


final EventDataBatch batch = currentBatch.getAndSet(null);
return producer.send(batch);
}
final EventDataBatch batch = currentBatch.getAndSet(null);
return producer.send(batch)
.doFinally(s -> producer.close());
}

private CreateBatchOptions buildCreateBatchOptions(PartitionSupplier partitionSupplier) {
Expand Down
10 changes: 5 additions & 5 deletions ...rvicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusTemplate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,11 +41,11 @@ public ServiceBusTemplate(@NonNull ServiceBusProducerFactory producerFactory) {
@Override
public <U> Mono<Void> sendAsync(String destination, Message<U> message) {
Assert.hasText(destination, "destination can't be null or empty");
try (ServiceBusSenderAsyncClient senderAsyncClient =
this.producerFactory.createProducer(destination, defaultEntityType)) {
ServiceBusMessage serviceBusMessage = messageConverter.fromMessage(message, ServiceBusMessage.class);
return senderAsyncClient.sendMessage(serviceBusMessage);
}
ServiceBusSenderAsyncClient senderAsyncClient =
this.producerFactory.createProducer(destination, defaultEntityType);
ServiceBusMessage serviceBusMessage = messageConverter.fromMessage(message, ServiceBusMessage.class);
return senderAsyncClient.sendMessage(serviceBusMessage)
.doFinally(t -> senderAsyncClient.close());
}

/**
Expand Down