-
Notifications
You must be signed in to change notification settings - Fork 352
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* provides extra hooks to ensure we capture all discarded elements Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provides leaks tracking tooling Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provides leaks tracking tests and tooling Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * more tests Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provides mechanism for terminates queue on calling clear Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provides workaround for FluxPublishOn to ensure that all elements are released in case of racing Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provides more tests part of the tests are on racing (ignored for now) another few on verification that elements are discarded properly Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * provide fixes to RequestChannel responder and related tests. Ensures there is no leaks in RSocketRequesterTest and RSocketResponder tests Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * tries to migrate to junit 5 Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * fixes leaks in tests Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * optimizes discarded/dropped BB consumption and releasing Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * fixes javadocs Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * removes hooks from Decoder Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * fixes format Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * rollbacks some fixes that should be delivered separately Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * rollbacks some build.gradle refactoring Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * fixes test Signed-off-by: Oleh Dokuka <shadowgun@i.ua> * fixes test Signed-off-by: Oleh Dokuka <shadowgun@i.ua>
- Loading branch information
1 parent
c05eb42
commit 070cffe
Showing
8 changed files
with
1,049 additions
and
102 deletions.
There are no files selected for viewing
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
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
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
167 changes: 167 additions & 0 deletions
167
rsocket-core/src/test/java/io/rsocket/buffer/LeaksTrackingByteBufAllocator.java
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| package io.rsocket.buffer; | ||
|
|
||
| import io.netty.buffer.ByteBuf; | ||
| import io.netty.buffer.ByteBufAllocator; | ||
| import io.netty.buffer.CompositeByteBuf; | ||
| import java.util.List; | ||
| import java.util.concurrent.ConcurrentLinkedQueue; | ||
| import org.assertj.core.api.Assertions; | ||
|
|
||
| /** | ||
| * Additional Utils which allows to decorate a ByteBufAllocator and track/assertOnLeaks all created | ||
| * ByteBuffs | ||
| */ | ||
| public class LeaksTrackingByteBufAllocator implements ByteBufAllocator { | ||
|
|
||
| /** | ||
| * Allows to instrument any given the instance of ByteBufAllocator | ||
| * | ||
| * @param allocator | ||
| * @return | ||
| */ | ||
| public static LeaksTrackingByteBufAllocator instrument(ByteBufAllocator allocator) { | ||
| return new LeaksTrackingByteBufAllocator(allocator); | ||
| } | ||
|
|
||
| final ConcurrentLinkedQueue<ByteBuf> tracker = new ConcurrentLinkedQueue<>(); | ||
|
|
||
| final ByteBufAllocator delegate; | ||
|
|
||
| private LeaksTrackingByteBufAllocator(ByteBufAllocator delegate) { | ||
| this.delegate = delegate; | ||
| } | ||
|
|
||
| public LeaksTrackingByteBufAllocator assertHasNoLeaks() { | ||
| try { | ||
| Assertions.assertThat(tracker) | ||
| .allSatisfy( | ||
| buf -> { | ||
| if (buf instanceof CompositeByteBuf) { | ||
| if (buf.refCnt() > 0) { | ||
| List<ByteBuf> decomposed = | ||
| ((CompositeByteBuf) buf).decompose(0, buf.readableBytes()); | ||
| for (int i = 0; i < decomposed.size(); i++) { | ||
| Assertions.assertThat(decomposed.get(i)) | ||
| .matches(bb -> bb.refCnt() == 0, "Got unreleased CompositeByteBuf"); | ||
| } | ||
| } | ||
|
|
||
| } else { | ||
| Assertions.assertThat(buf) | ||
| .matches(bb -> bb.refCnt() == 0, "buffer should be released"); | ||
| } | ||
| }); | ||
| } finally { | ||
| tracker.clear(); | ||
| } | ||
| return this; | ||
| } | ||
|
|
||
| // Delegating logic with tracking of buffers | ||
|
|
||
| @Override | ||
| public ByteBuf buffer() { | ||
| return track(delegate.buffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf buffer(int initialCapacity) { | ||
| return track(delegate.buffer(initialCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf buffer(int initialCapacity, int maxCapacity) { | ||
| return track(delegate.buffer(initialCapacity, maxCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf ioBuffer() { | ||
| return track(delegate.ioBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf ioBuffer(int initialCapacity) { | ||
| return track(delegate.ioBuffer(initialCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf ioBuffer(int initialCapacity, int maxCapacity) { | ||
| return track(delegate.ioBuffer(initialCapacity, maxCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf heapBuffer() { | ||
| return track(delegate.heapBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf heapBuffer(int initialCapacity) { | ||
| return track(delegate.heapBuffer(initialCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf heapBuffer(int initialCapacity, int maxCapacity) { | ||
| return track(delegate.heapBuffer(initialCapacity, maxCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf directBuffer() { | ||
| return track(delegate.directBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf directBuffer(int initialCapacity) { | ||
| return track(delegate.directBuffer(initialCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public ByteBuf directBuffer(int initialCapacity, int maxCapacity) { | ||
| return track(delegate.directBuffer(initialCapacity, maxCapacity)); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeBuffer() { | ||
| return track(delegate.compositeBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeBuffer(int maxNumComponents) { | ||
| return track(delegate.compositeBuffer(maxNumComponents)); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeHeapBuffer() { | ||
| return track(delegate.compositeHeapBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeHeapBuffer(int maxNumComponents) { | ||
| return track(delegate.compositeHeapBuffer(maxNumComponents)); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeDirectBuffer() { | ||
| return track(delegate.compositeDirectBuffer()); | ||
| } | ||
|
|
||
| @Override | ||
| public CompositeByteBuf compositeDirectBuffer(int maxNumComponents) { | ||
| return track(delegate.compositeDirectBuffer(maxNumComponents)); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDirectBufferPooled() { | ||
| return delegate.isDirectBufferPooled(); | ||
| } | ||
|
|
||
| @Override | ||
| public int calculateNewCapacity(int minNewCapacity, int maxCapacity) { | ||
| return delegate.calculateNewCapacity(minNewCapacity, maxCapacity); | ||
| } | ||
|
|
||
| <T extends ByteBuf> T track(T buffer) { | ||
| tracker.offer(buffer); | ||
|
|
||
| return buffer; | ||
| } | ||
| } |
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
Oops, something went wrong.