diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/buffer/DataBuf.java b/driver/src/main/java/eu/cloudnetservice/driver/network/buffer/DataBuf.java index d91e8326e4..3088403b26 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/buffer/DataBuf.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/buffer/DataBuf.java @@ -231,8 +231,8 @@ public interface DataBuf extends AutoCloseable { T readObject(@NonNull Type type); /** - * Reads the next requested data from the buffer. This method call is equivalent to {@code - * readNullable(readerWhenNonNull, null)}. + * Reads the next requested data from the buffer. This method call is equivalent to + * {@code readNullable(readerWhenNonNull, null)}. * * @param readerWhenNonNull the reader to read the data from the buffer when the next value is non-null. * @param the generic type of the data to read. @@ -300,27 +300,32 @@ public interface DataBuf extends AutoCloseable { boolean accessible(); /** - * Disables that the underlying buffer can be released when the reader index of this buffer reaches the tail or an - * explicit call to {@code release()} or {@code close()} is made. + * Get the amount of acquires that this data buf has. Initially a data buf is acquired once. * - * @return the same instance as used to call the method, for chaining. + * @return the amount of acquires. A value smaller or equal to zero means that the buffer was released. */ - @NonNull DataBuf disableReleasing(); + int acquires(); /** - * Enables that the underlying buffer can be released when the reader index of this buffer reaches the tail or an - * explicit call to {@code release()} or {@code close()} is made. + * Acquires this buffer once. If a buffer gets acquired, further calls to {@link #release()} will decrease the count, + * but only release the buffer if there were more release than acquire calls. * * @return the same instance as used to call the method, for chaining. */ - @NonNull DataBuf enableReleasing(); + @NonNull DataBuf acquire(); /** - * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does - * nothing if releasing was disables before calling this method. + * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method only + * decreases the acquire count of the buffer in case it was acquired at least once. */ void release(); + /** + * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does + * not check if anyone acquired the buffer, it will be released in any case. + */ + void forceRelease(); + /** * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does * nothing if releasing was disables before calling this method. diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/DefaultFileChunkPacketSender.java b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/DefaultFileChunkPacketSender.java index 72bed981f7..111d0b6593 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/DefaultFileChunkPacketSender.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/DefaultFileChunkPacketSender.java @@ -88,8 +88,12 @@ public DefaultFileChunkPacketSender( while (true) { var bytesRead = this.source.read(backingArray); if (bytesRead != -1 && bytesRead == backingArray.length) { - this.packetSplitter.accept( - ChunkedPacket.createChunk(this.chunkSessionInformation, readCalls++, backingArray)); + // acquire the transfer information once before writing the data of the chunk + this.chunkSessionInformation.transferInformation().acquire(); + this.packetSplitter.accept(ChunkedPacket.createChunk( + this.chunkSessionInformation, + readCalls++, + backingArray)); } else { this.packetSplitter.accept(ChunkedPacket.createChunk( this.chunkSessionInformation, @@ -97,10 +101,11 @@ public DefaultFileChunkPacketSender( readCalls, bytesRead == -1 ? 0 : bytesRead, bytesRead == -1 ? EMPTY_BYTE_ARRAY : backingArray)); - // close the stream after reading the final chunk + + // close the stream after reading the final chunk & release the extra content now this.source.close(); - // release the extra content now - this.chunkSessionInformation.transferInformation().enableReleasing().release(); + this.chunkSessionInformation.transferInformation().release(); + // successful transfer return TransferStatus.SUCCESS; } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/builder/DefaultChunkedPacketSenderBuilder.java b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/builder/DefaultChunkedPacketSenderBuilder.java index d4bee84b4e..f26de55b49 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/builder/DefaultChunkedPacketSenderBuilder.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/builder/DefaultChunkedPacketSenderBuilder.java @@ -115,7 +115,9 @@ public abstract class DefaultChunkedPacketSenderBuilder implements ChunkedPacket */ @Override public @NonNull ChunkedPacketSender.Builder withExtraData(@NonNull DataBuf extraData) { - this.transferInformation = extraData.disableReleasing(); + // make sure that the old transfer information buffer is released + this.transferInformation.release(); + this.transferInformation = extraData; return this; } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/splitter/NetworkChannelsPacketSplitter.java b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/splitter/NetworkChannelsPacketSplitter.java index 0a07021a8e..86516220af 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/splitter/NetworkChannelsPacketSplitter.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/splitter/NetworkChannelsPacketSplitter.java @@ -37,18 +37,19 @@ public record NetworkChannelsPacketSplitter(@NonNull Collection */ @Override public void accept(@NonNull Packet packet) { - // disable releasing of the content as we need to content multiple times - packet.content().disableReleasing(); // write to all channels for (var channel : this.channels) { - // mark the current indexes of the packet - packet.content().startTransaction(); - // write the packet to the channel - channel.sendPacketSync(packet); - // redo the transaction to allow further writes - packet.content().redoTransaction(); + try { + // acquire the packet content and start a transaction, then send the data + packet.content().acquire().startTransaction(); + channel.sendPacketSync(packet); + } finally { + // redo the transaction to allow further writes + packet.content().redoTransaction(); + } } - // enable the releasing and free the memory - packet.content().enableReleasing().release(); + + // release the packet content + packet.content().release(); } } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/NettyNetworkHandler.java b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/NettyNetworkHandler.java index 2f35d4c268..ace97c446f 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/NettyNetworkHandler.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/NettyNetworkHandler.java @@ -109,16 +109,23 @@ protected void doHandlePacket(@NonNull BasePacket packet) { if (uuid != null) { var task = this.channel.queryPacketManager().waitingHandler(uuid); if (task != null) { + // complete the waiting task task.complete(packet); + // don't post a query response packet to another handler at all + // the packet might be inbound - we might be expected to respond return; } } - // check if we're allowed to handle the packet - if (this.channel.handler().handlePacketReceive(this.channel, packet)) { - this.channel.packetRegistry().handlePacket(this.channel, packet); + // check if any handler can handle the incoming packet + if (this.channel.handler().handlePacketReceive(this.channel, packet) + && this.channel.packetRegistry().handlePacket(this.channel, packet)) { + return; } + + // release the packet content now, there are no handlers that are accepting the message + packet.content().forceRelease(); } catch (Exception exception) { LOGGER.severe("Exception whilst handling packet %s", exception, packet); } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyImmutableDataBuf.java b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyImmutableDataBuf.java index 0d0f67cb52..3a2e60bc51 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyImmutableDataBuf.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyImmutableDataBuf.java @@ -36,11 +36,13 @@ public class NettyImmutableDataBuf implements DataBuf { protected final Buffer buffer; - protected boolean releasable; + + // the amount of times this buffer was acquired + protected int acquires = 1; // transaction offset data - protected volatile int readOffset; - protected volatile int writeOffset; + protected int readOffset; + protected int writeOffset; /** * Constructs a new netty immutable data buf instance. @@ -50,7 +52,6 @@ public class NettyImmutableDataBuf implements DataBuf { */ public NettyImmutableDataBuf(@NonNull Buffer buffer) { this.buffer = buffer; - this.enableReleasing(); } /** @@ -205,10 +206,8 @@ public T readObject(@NonNull Type type) { */ @Override public T readNullable(@NonNull Function readerWhenNonNull, T valueWhenNull) { - return this.hotRead(buf -> { - var isNonNull = buf.readBoolean(); - return isNonNull ? readerWhenNonNull.apply(this) : valueWhenNull; - }); + var isNonNull = this.readBoolean(); + return isNonNull ? readerWhenNonNull.apply(this) : valueWhenNull; } /** @@ -265,17 +264,16 @@ public boolean accessible() { * {@inheritDoc} */ @Override - public @NonNull DataBuf disableReleasing() { - this.releasable = false; - return this; + public int acquires() { + return this.acquires; } /** * {@inheritDoc} */ @Override - public @NonNull DataBuf enableReleasing() { - this.releasable = true; + public @NonNull DataBuf acquire() { + this.acquires++; return this; } @@ -284,7 +282,25 @@ public boolean accessible() { */ @Override public void release() { - if (this.releasable && this.buffer.isAccessible()) { + // release one acquire + this.acquires--; + + // check if the buffer is no longer acquired somewhere + if (this.acquires <= 0 && this.buffer.isAccessible()) { + this.buffer.close(); + } + } + + /** + * {@inheritDoc} + */ + @Override + public void forceRelease() { + // set acquires to 0 to indicate that the buffer was released + this.acquires = 0; + + // actually release the buffer if needed + if (this.buffer.isAccessible()) { this.buffer.close(); } } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyMutableDataBuf.java b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyMutableDataBuf.java index 58980193e1..aba3453085 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyMutableDataBuf.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyMutableDataBuf.java @@ -158,11 +158,14 @@ public NettyMutableDataBuf(@NonNull Buffer buffer) { @Override public @NonNull DataBuf.Mutable writeDataBuf(@NonNull DataBuf buf) { buf.startTransaction(); - // write the content - this.writeInt(buf.readableBytes()); - this.buffer.writeBytes(((NettyImmutableDataBuf) buf).buffer); - // reset the data for later use - buf.redoTransaction(); + try { + // write the content + this.writeInt(buf.readableBytes()); + this.buffer.writeBytes(((NettyImmutableDataBuf) buf).buffer); + } finally { + // reset the data for later use & release the content if it isn't disabled + buf.redoTransaction().release(); + } return this; } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/PacketListenerRegistry.java b/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/PacketListenerRegistry.java index 2c0c4cb33c..f6c6ba6f49 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/PacketListenerRegistry.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/PacketListenerRegistry.java @@ -138,7 +138,8 @@ public interface PacketListenerRegistry { * * @param channel the channel from which the packet came. * @param packet the packet which was received. + * @return true if a listener handled the packet, false otherwise. * @throws NullPointerException if either the given channel or packet is null. */ - void handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet); + boolean handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet); } diff --git a/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/defaults/DefaultPacketListenerRegistry.java b/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/defaults/DefaultPacketListenerRegistry.java index 42e49a850e..3a8c77c206 100644 --- a/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/defaults/DefaultPacketListenerRegistry.java +++ b/driver/src/main/java/eu/cloudnetservice/driver/network/protocol/defaults/DefaultPacketListenerRegistry.java @@ -160,23 +160,29 @@ public void removeListeners() { * {@inheritDoc} */ @Override - public void handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet) { - if (this.parent != null) { - this.parent.handlePacket(channel, packet); - } + public boolean handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet) { + // check if the parent handler registry is present and can handle the packet + var parentDidHandle = this.parent != null && this.parent.handlePacket(channel, packet); + // get the listeners that are registered in this packet registry for the packet channel var registeredListeners = this.listeners.get(packet.channel()); - if (!registeredListeners.isEmpty()) { - for (var listener : registeredListeners) { - try { - listener.handle(channel, packet); - } catch (Exception exception) { - throw new IllegalStateException(String.format( - "Exception posting packet from channel %d to handler %s", - packet.channel(), listener.getClass().getCanonicalName() - ), exception); - } + if (registeredListeners.isEmpty()) { + return parentDidHandle; + } + + // post the packet to each listener that handles the packet + for (var listener : registeredListeners) { + try { + listener.handle(channel, packet); + } catch (Exception exception) { + throw new IllegalStateException(String.format( + "Exception posting packet from channel %d to handler %s", + packet.channel(), listener.getClass().getCanonicalName() + ), exception); } } + + // at least one listener handled the packet + return true; } } diff --git a/node/src/main/java/eu/cloudnetservice/node/cluster/sync/DefaultDataSyncRegistry.java b/node/src/main/java/eu/cloudnetservice/node/cluster/sync/DefaultDataSyncRegistry.java index 922a582998..09d55dcb78 100644 --- a/node/src/main/java/eu/cloudnetservice/node/cluster/sync/DefaultDataSyncRegistry.java +++ b/node/src/main/java/eu/cloudnetservice/node/cluster/sync/DefaultDataSyncRegistry.java @@ -188,8 +188,7 @@ public boolean hasHandler(@NonNull String handlerKey) { // no handler for the result LOGGER.fine("No handler for key %s to sync data", null, key); } - // try to release the input buf - input.release(); + // return the created result return result; } diff --git a/node/src/main/java/eu/cloudnetservice/node/command/sub/ClusterCommand.java b/node/src/main/java/eu/cloudnetservice/node/command/sub/ClusterCommand.java index 0556af477f..139d148817 100644 --- a/node/src/main/java/eu/cloudnetservice/node/command/sub/ClusterCommand.java +++ b/node/src/main/java/eu/cloudnetservice/node/command/sub/ClusterCommand.java @@ -352,16 +352,20 @@ private void pushTemplate(@NonNull CommandSource source, @NonNull ServiceTemplat // check if the template really exists in the given storage if (inputStream != null) { // deploy the template into the cluster - this.nodeServerProvider.deployTemplateToCluster(template, inputStream, true) - .thenAccept(transferStatus -> { - if (transferStatus == TransferStatus.FAILURE) { - // the transfer failed - source.sendMessage(I18n.trans("command-cluster-push-template-failed", templateName)); - } else { - // the transfer was successful - source.sendMessage(I18n.trans("command-cluster-push-template-success", templateName)); + this.nodeServerProvider.deployTemplateToCluster(template, inputStream, true).whenComplete((status, ex) -> { + if (ex != null || status == TransferStatus.FAILURE) { + // the transfer failed + source.sendMessage(I18n.trans("command-cluster-push-template-failed", templateName)); + + // print the detailed exception, if available + if (ex != null) { + LOGGER.severe("Unable to push template %s to cluster", ex, template); } - }); + } else { + // the transfer was successful + source.sendMessage(I18n.trans("command-cluster-push-template-success", templateName)); + } + }); } else { source.sendMessage(I18n.trans("command-template-not-found", templateName)); } diff --git a/node/src/main/java/eu/cloudnetservice/node/network/listener/PacketServerChannelMessageListener.java b/node/src/main/java/eu/cloudnetservice/node/network/listener/PacketServerChannelMessageListener.java index 53bf036395..ec4eb64f04 100644 --- a/node/src/main/java/eu/cloudnetservice/node/network/listener/PacketServerChannelMessageListener.java +++ b/node/src/main/java/eu/cloudnetservice/node/network/listener/PacketServerChannelMessageListener.java @@ -54,29 +54,31 @@ public PacketServerChannelMessageListener( public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { var comesFromWrapper = packet.content().readBoolean(); var message = packet.content().readObject(ChannelMessage.class); - // disable releasing of the message content - message.content().disableReleasing(); + // check if we should handle the message locally var handleLocally = message.targets().stream().anyMatch(target -> switch (target.type()) { case ALL -> true; case NODE -> target.name() == null || target.name().equals(this.componentInfo.componentName()); default -> false; }); + if (handleLocally) { - // mark the index of the data buf - message.content().startTransaction(); + // mark the index of the data buf & call the receive event + message.content().acquire().startTransaction(); // call the receive event var responseTask = this.eventManager .callEvent(new ChannelMessageReceiveEvent(message, channel, packet.uniqueId() != null)) .queryResponse(); // reset the index message.content().redoTransaction(); + // wait for the response to become available if given before resuming if (responseTask != null) { responseTask.thenAccept(response -> this.resumeHandling(packet, channel, message, response, comesFromWrapper)); return; } } + // resume instantly this.resumeHandling(packet, channel, message, null, comesFromWrapper); } @@ -93,8 +95,8 @@ private void resumeHandling( this.messenger.sendChannelMessageQueryAsync(message, comesFromWrapper) .orTimeout(20, TimeUnit.SECONDS) .whenComplete((result, exception) -> { - DataBuf responseContent; // check if the handling was successful + DataBuf responseContent; if (exception == null) { // respond with the result or just the single initial response if given if (result == null) { @@ -106,6 +108,7 @@ private void resumeHandling( if (initialResponse != null) { result.add(initialResponse); } + // serialize the response if (result.isEmpty()) { responseContent = DataBuf.empty().writeBoolean(false); @@ -117,13 +120,27 @@ private void resumeHandling( // just respond with nothing when an exception was thrown responseContent = DataBuf.empty().writeBoolean(false); } + // send the results to the sender channel.sendPacket(packet.constructResponse(responseContent)); }); } else { this.messenger.sendChannelMessage(message, comesFromWrapper); } - // force release the message - message.content().enableReleasing().release(); + + // release the initial response content in case it was not send to any channel + if (initialResponse != null) { + initialResponse.content().release(); + } + + // force release of the current message + // this is an edge case that should not happen, but basically the handlers did not read + // all the content from the buffer, or the buffer simply had no content + // checking if the buffer was acquired only once ensures that no-one acquired the buffer + // during the read process and wants to use the buffer later on + var messageContent = message.content(); + if (messageContent.acquires() == 1) { + messageContent.release(); + } } } diff --git a/node/src/main/java/eu/cloudnetservice/node/provider/NodeMessenger.java b/node/src/main/java/eu/cloudnetservice/node/provider/NodeMessenger.java index 96e27dcc9f..5daa9f1d87 100644 --- a/node/src/main/java/eu/cloudnetservice/node/provider/NodeMessenger.java +++ b/node/src/main/java/eu/cloudnetservice/node/provider/NodeMessenger.java @@ -77,38 +77,67 @@ public void sendChannelMessage(@NonNull ChannelMessage message) { } public void sendChannelMessage(@NonNull ChannelMessage message, boolean allowClusterRedirect) { - for (var channel : this.findChannels(message.targets(), allowClusterRedirect)) { + // find the target channels to send the message to + var channels = this.findChannels(message.targets(), allowClusterRedirect); + for (var channel : channels) { + // acquire the message content for each channel we're sending the message to + // when writing the message content to the target buffer, the message is released + // that means when the message was written to all channels it's released unless someone acquired it before + message.content().acquire(); + + // construct and send the packet + var packet = new PacketServerChannelMessage(message, false); if (message.sendSync()) { - channel.sendPacketSync(new PacketServerChannelMessage(message, false)); + channel.sendPacketSync(packet); } else { - channel.sendPacket(new PacketServerChannelMessage(message, false)); + channel.sendPacket(packet); } } + + // release the message now + message.content().release(); } public @NonNull Task> sendChannelMessageQueryAsync( @NonNull ChannelMessage message, boolean allowClusterRedirect ) { - // filter the channels we need to send the message to + // find the target channels to send the message to var channels = this.findChannels(message.targets(), allowClusterRedirect); - // the result we generate - Set result = new HashSet<>(); - var task = new CountingTask>(result, channels.size()); - // send the packet to each channel - for (var channel : channels) { - channel.sendQueryAsync(new PacketServerChannelMessage(message, false)).whenComplete((packet, th) -> { - // check if we got an actual result from the request - if (th == null && packet.readable()) { - // add all resulting messages we got - result.addAll(packet.content().readObject(COL_MSG)); - } - // count down - one channel responded - task.countDown(); - }); + if (channels.isEmpty()) { + // no target channels found, release the message now + message.content().release(); + return Task.completedTask(new HashSet<>()); + } else { + // the result we generate + Set result = new HashSet<>(); + var task = new CountingTask>(result, channels.size()); + + // send the packet to each channel + for (var channel : channels) { + // acquire the message content for each channel we're sending the message to + // when writing the message content to the target buffer, the message is released + // that means when the message was written to all channels it's released unless someone acquired it before + message.content().acquire(); + + channel.sendQueryAsync(new PacketServerChannelMessage(message, false)).whenComplete((packet, th) -> { + // check if we got an actual result from the request + if (th == null && packet.readable()) { + // add all resulting messages we got + result.addAll(packet.content().readObject(COL_MSG)); + } + + // count down - one channel responded + task.countDown(); + }); + } + + // release the message now + message.content().release(); + + // return the task on which the user can wait + return task; } - // return the task on which the user can wait - return task; } protected @NonNull Collection findChannels( diff --git a/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketAuthorizationResponseListener.java b/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketAuthorizationResponseListener.java index 96e6371f84..70a367892d 100644 --- a/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketAuthorizationResponseListener.java +++ b/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketAuthorizationResponseListener.java @@ -36,7 +36,12 @@ public PacketAuthorizationResponseListener(@NonNull Thread blockedThread) { @Override public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { // read the auth result - this.result.setRelease(packet.content().readBoolean()); + var content = packet.content(); + this.result.setRelease(content.readBoolean()); + + // skip the next two booleans from the packet + content.readBoolean(); + content.readBoolean(); // signal all listeners waiting for the auth LockSupport.unpark(this.blockedThread); diff --git a/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketServerChannelMessageListener.java b/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketServerChannelMessageListener.java index c22397f5cf..badc07e348 100644 --- a/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketServerChannelMessageListener.java +++ b/wrapper-jvm/src/main/java/eu/cloudnetservice/wrapper/network/listener/PacketServerChannelMessageListener.java @@ -24,9 +24,11 @@ import eu.cloudnetservice.driver.network.protocol.Packet; import eu.cloudnetservice.driver.network.protocol.PacketListener; import jakarta.inject.Inject; +import jakarta.inject.Singleton; import java.util.Set; import lombok.NonNull; +@Singleton public final class PacketServerChannelMessageListener implements PacketListener { private final EventManager eventManager; @@ -38,14 +40,16 @@ public PacketServerChannelMessageListener(@NonNull EventManager eventManager) { @Override public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { - // skip the first boolean in the buffer as we don't need it + // skip the first boolean (comes from wrapper) in the buffer as we don't need it packet.content().readBoolean(); // read the channel message from the buffer var message = packet.content().readObject(ChannelMessage.class); + // get the query response if available var response = this.eventManager .callEvent(new ChannelMessageReceiveEvent(message, channel, packet.uniqueId() != null)) .queryResponse(); + // check if we need to respond to the channel message if (packet.uniqueId() != null) { // wait for the future if a response was supplied @@ -58,11 +62,27 @@ public void handle(@NonNull NetworkChannel channel, @NonNull Packet packet) { // serialize the single response channel.sendPacket(packet.constructResponse(DataBuf.empty().writeObject(Set.of(queryResponse)))); } + + // release the message content (do it here so that the async processing still has access to it) + message.content().release(); }); } else { - // respond with an empty buffer to indicate the node that there was no result + // respond with an empty buffer to indicate the node that there was no result & release the message content channel.sendPacket(packet.constructResponse(DataBuf.empty())); + message.content().release(); } + } else if (response != null) { + // just release the initial response content when available + response.thenAccept(responseMessage -> { + // release both messages + message.content().release(); + if (responseMessage != null) { + responseMessage.content().release(); + } + }); + } else { + // release the message content instantly + message.content().release(); } } }