99
1010import static org .junit .Assert .*;
1111
12+ import java .net .URI ;
1213import java .util .Arrays ;
1314import java .util .Map ;
1415import org .junit .Test ;
1516
1617public class BoundedQueueTest {
18+ private static final URI URL = URI .create ("http://example.invalid/" );
19+
1720 private static byte [] bytes (int n ) {
1821 return bytes (n , (byte ) 0 );
1922 }
@@ -24,24 +27,32 @@ private static byte[] bytes(int n, byte v) {
2427 return b ;
2528 }
2629
30+ private static Payload payload (int n ) {
31+ return new Payload (URL , bytes (n ));
32+ }
33+
34+ private static Payload payload (byte [] b ) {
35+ return new Payload (URL , b );
36+ }
37+
2738 // --- Round-trip / bytes tracking ---
2839
2940 @ Test
3041 public void addThenNextReturnsItem () throws InterruptedException {
3142 BoundedQueue q = new BoundedQueue (10 , 1 , WhenFull .DROP );
32- byte [] item = bytes (4 );
43+ Payload item = payload (4 );
3344 q .add (item );
34- Map .Entry <BoundedQueue .Key , byte [] > entry = q .next ();
45+ Map .Entry <BoundedQueue .Key , Payload > entry = q .next ();
3546 assertSame (item , entry .getValue ());
3647 assertEquals (0 , q .bytes );
3748 }
3849
3950 @ Test
4051 public void bytesDecrementedOnNext () throws InterruptedException {
4152 BoundedQueue q = new BoundedQueue (30 , 1 , WhenFull .DROP );
42- q .add (bytes (3 ));
43- q .add (bytes (3 ));
44- q .add (bytes (3 ));
53+ q .add (payload (3 ));
54+ q .add (payload (3 ));
55+ q .add (payload (3 ));
4556 assertEquals (9 , q .bytes );
4657 q .next ();
4758 assertEquals (6 , q .bytes );
@@ -54,9 +65,9 @@ public void bytesDecrementedOnNext() throws InterruptedException {
5465 @ Test
5566 public void newestItemDequeuesFirstWithinSameTries () throws InterruptedException {
5667 BoundedQueue q = new BoundedQueue (30 , 1 , WhenFull .DROP );
57- byte [] a = {1 };
58- byte [] b = {2 };
59- byte [] c = {3 };
68+ Payload a = payload ( new byte [] {1 }) ;
69+ Payload b = payload ( new byte [] {2 }) ;
70+ Payload c = payload ( new byte [] {3 }) ;
6071 q .add (a );
6172 q .add (b );
6273 q .add (c );
@@ -71,11 +82,11 @@ public void newestItemDequeuesFirstWithinSameTries() throws InterruptedException
7182 @ Test
7283 public void dropWhenFullDropsOldestItem () throws InterruptedException {
7384 BoundedQueue q = new BoundedQueue (10 , 1 , WhenFull .DROP );
74- byte [] a = bytes (5 ); // added first → smallest clock → last in TreeMap
75- byte [] b = bytes (4 );
85+ Payload a = payload (5 ); // added first → smallest clock → last in TreeMap
86+ Payload b = payload (4 );
7687 q .add (a ); // queue: a(clock=MIN+1)
7788 q .add (b ); // queue full: a, b
78- byte [] c = bytes (3 );
89+ Payload c = payload (3 );
7990 q .add (c ); // a (oldest, last entry) evicted
8091 assertEquals (1 , q .droppedItems );
8192 assertEquals (5 , q .droppedBytes );
@@ -87,32 +98,32 @@ public void dropWhenFullDropsOldestItem() throws InterruptedException {
8798 @ Test
8899 public void dropCountersAccumulate () throws InterruptedException {
89100 BoundedQueue q = new BoundedQueue (5 , 1 , WhenFull .DROP );
90- q .add (bytes (5 )); // fills queue (X)
91- q .add (bytes (5 )); // X dropped (Y in)
92- q .add (bytes (5 )); // Y dropped (Z in)
101+ q .add (payload (5 )); // fills queue (X)
102+ q .add (payload (5 )); // X dropped (Y in)
103+ q .add (payload (5 )); // Y dropped (Z in)
93104 assertEquals (2 , q .droppedItems );
94105 assertEquals (10 , q .droppedBytes );
95106 }
96107
97108 @ Test (timeout = 3000 )
98109 public void dropDoesNotBlock () throws InterruptedException {
99110 BoundedQueue q = new BoundedQueue (5 , 1 , WhenFull .DROP );
100- q .add (bytes (5 )); // fill
101- q .add (bytes (5 )); // should return immediately via DROP
111+ q .add (payload (5 )); // fill
112+ q .add (payload (5 )); // should return immediately via DROP
102113 }
103114
104115 // --- WhenFull.BLOCK ---
105116
106117 @ Test (timeout = 5000 )
107118 public void blockUnblocksWhenSpaceFreed () throws InterruptedException {
108119 BoundedQueue q = new BoundedQueue (5 , 1 , WhenFull .BLOCK );
109- q .add (bytes (5 )); // queue full
120+ q .add (payload (5 )); // queue full
110121
111122 Thread producer =
112123 new Thread (
113124 () -> {
114125 try {
115- q .add (bytes (5 ));
126+ q .add (payload (5 ));
116127 } catch (InterruptedException e ) {
117128 return ;
118129 }
@@ -135,19 +146,19 @@ public void blockUnblocksWhenSpaceFreed() throws InterruptedException {
135146 @ Test (expected = IllegalArgumentException .class )
136147 public void addThrowsForOversizedItem () throws InterruptedException {
137148 BoundedQueue q = new BoundedQueue (4 , 1 , WhenFull .DROP );
138- q .add (bytes (5 ));
149+ q .add (payload (5 ));
139150 }
140151
141152 @ Test
142153 public void requeueIncrementsTriesPreservesClock () throws InterruptedException {
143154 BoundedQueue q = new BoundedQueue (20 , 3 , WhenFull .DROP );
144- q .add (bytes (4 ));
145- Map .Entry <BoundedQueue .Key , byte [] > entry = q .next ();
155+ q .add (payload (4 ));
156+ Map .Entry <BoundedQueue .Key , Payload > entry = q .next ();
146157 assertEquals (0 , entry .getKey ().tries );
147158 long originalClock = entry .getKey ().clock ;
148159
149160 q .requeue (entry );
150- Map .Entry <BoundedQueue .Key , byte [] > requeued = q .next ();
161+ Map .Entry <BoundedQueue .Key , Payload > requeued = q .next ();
151162 assertEquals (1 , requeued .getKey ().tries );
152163 assertEquals (originalClock , requeued .getKey ().clock );
153164 assertEquals (0 , q .droppedItems );
@@ -156,10 +167,10 @@ public void requeueIncrementsTriesPreservesClock() throws InterruptedException {
156167 @ Test
157168 public void requeuedItemDequeuesAfterFreshItems () throws InterruptedException {
158169 BoundedQueue q = new BoundedQueue (20 , 3 , WhenFull .DROP );
159- byte [] a = {10 };
160- byte [] b = {20 };
170+ Payload a = payload ( new byte [] {10 }) ;
171+ Payload b = payload ( new byte [] {20 }) ;
161172 q .add (a );
162- Map .Entry <BoundedQueue .Key , byte [] > entryA = q .next ();
173+ Map .Entry <BoundedQueue .Key , Payload > entryA = q .next ();
163174 q .requeue (entryA ); // A now has tries=1
164175
165176 q .add (b ); // B has tries=0 → higher priority
@@ -171,8 +182,8 @@ public void requeuedItemDequeuesAfterFreshItems() throws InterruptedException {
171182 @ Test
172183 public void requeueAtMaxTriesIsAccepted () throws InterruptedException {
173184 BoundedQueue q = new BoundedQueue (20 , 2 , WhenFull .DROP );
174- q .add (bytes (3 ));
175- Map .Entry <BoundedQueue .Key , byte [] > e = q .next ();
185+ q .add (payload (3 ));
186+ Map .Entry <BoundedQueue .Key , Payload > e = q .next ();
176187 q .requeue (e ); // tries → 1
177188 e = q .next ();
178189 q .requeue (e ); // tries → 2 == maxTries, should be accepted
@@ -183,9 +194,8 @@ public void requeueAtMaxTriesIsAccepted() throws InterruptedException {
183194 @ Test
184195 public void requeuePastMaxTriesDropsItem () throws InterruptedException {
185196 BoundedQueue q = new BoundedQueue (20 , 2 , WhenFull .DROP );
186- byte [] item = bytes (7 );
187- q .add (item );
188- Map .Entry <BoundedQueue .Key , byte []> e = q .next ();
197+ q .add (payload (7 ));
198+ Map .Entry <BoundedQueue .Key , Payload > e = q .next ();
189199 q .requeue (e ); // tries → 1
190200 e = q .next ();
191201 q .requeue (e ); // tries → 2 == maxTries, accepted
@@ -200,8 +210,8 @@ public void requeuePastMaxTriesDropsItem() throws InterruptedException {
200210 @ Test (timeout = 5000 )
201211 public void nextBlocksUntilItemAdded () throws InterruptedException {
202212 BoundedQueue q = new BoundedQueue (100 , 1 , WhenFull .DROP );
203- byte [] item = bytes (3 );
204- Map .Entry <BoundedQueue .Key , byte [] >[] result = new Map .Entry [1 ];
213+ Payload item = payload (3 );
214+ Map .Entry <BoundedQueue .Key , Payload >[] result = new Map .Entry [1 ];
205215
206216 Thread consumer =
207217 new Thread (
0 commit comments