Skip to content

Commit 9a02d4d

Browse files
OlegDokukaOlegDokuka
committed
renames PooledRSocket into WeightedRSocket
Signed-off-by: Oleh Dokuka <[email protected]>
1 parent 56330fa commit 9a02d4d

File tree

6 files changed

+65
-68
lines changed

6 files changed

+65
-68
lines changed

‎rsocket-core/src/main/java/io/rsocket/loadbalance/LoadbalanceStrategy.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
@FunctionalInterface
2222
public interface LoadbalanceStrategy {
2323

24-
PooledRSocket select(List<PooledRSocket> availableRSockets);
24+
WeightedRSocket select(List<WeightedRSocket> availableRSockets);
2525

2626
default Supplier<Stats> statsSupplier() {
2727
return Stats::noOps;

‎rsocket-core/src/main/java/io/rsocket/loadbalance/DefaultPooledRSocket.java renamed to ‎rsocket-core/src/main/java/io/rsocket/loadbalance/PooledWeightedRSocket.java

Lines changed: 19 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -28,20 +28,20 @@
2828
import reactor.core.publisher.Operators;
2929
import reactor.util.context.Context;
3030

31-
/** Default implementation of {@link PooledRSocket} stored in {@link RSocketPool} */
32-
final class DefaultPooledRSocket extends ResolvingOperator<RSocket>
33-
implements CoreSubscriber<RSocket>, PooledRSocket {
31+
/** Default implementation of {@link WeightedRSocket} stored in {@link RSocketPool} */
32+
final class PooledWeightedRSocket extends ResolvingOperator<RSocket>
33+
implements CoreSubscriber<RSocket>, WeightedRSocket {
3434

3535
final RSocketPool parent;
3636
final LoadbalanceRSocketSource loadbalanceRSocketSource;
3737
final Stats stats;
3838

3939
volatile Subscription s;
4040

41-
static final AtomicReferenceFieldUpdater<DefaultPooledRSocket, Subscription> S =
42-
AtomicReferenceFieldUpdater.newUpdater(DefaultPooledRSocket.class, Subscription.class, "s");
41+
static final AtomicReferenceFieldUpdater<PooledWeightedRSocket, Subscription> S =
42+
AtomicReferenceFieldUpdater.newUpdater(PooledWeightedRSocket.class, Subscription.class, "s");
4343

44-
DefaultPooledRSocket(
44+
PooledWeightedRSocket(
4545
RSocketPool parent, LoadbalanceRSocketSource loadbalanceRSocketSource, Stats stats) {
4646
this.parent = parent;
4747
this.stats = stats;
@@ -128,7 +128,7 @@ public void dispose() {
128128
protected void doOnDispose() {
129129
final RSocketPool parent = this.parent;
130130
for (; ; ) {
131-
final PooledRSocket[] sockets = parent.activeSockets;
131+
final WeightedRSocket[] sockets = parent.activeSockets;
132132
final int activeSocketsCount = sockets.length;
133133

134134
int index = -1;
@@ -144,7 +144,7 @@ protected void doOnDispose() {
144144
}
145145

146146
final int lastIndex = activeSocketsCount - 1;
147-
final PooledRSocket[] newSockets = new PooledRSocket[lastIndex];
147+
final WeightedRSocket[] newSockets = new WeightedRSocket[lastIndex];
148148
if (index != 0) {
149149
System.arraycopy(sockets, 0, newSockets, 0, index);
150150
}
@@ -196,8 +196,7 @@ public Stats stats() {
196196
return stats;
197197
}
198198

199-
@Override
200-
public LoadbalanceRSocketSource source() {
199+
LoadbalanceRSocketSource source() {
201200
return loadbalanceRSocketSource;
202201
}
203202

@@ -211,7 +210,7 @@ static final class RequestTrackingMonoInner<RESULT>
211210

212211
long startTime;
213212

214-
RequestTrackingMonoInner(DefaultPooledRSocket parent, Payload payload, FrameType requestType) {
213+
RequestTrackingMonoInner(PooledWeightedRSocket parent, Payload payload, FrameType requestType) {
215214
super(parent, payload, requestType);
216215
}
217216

@@ -245,7 +244,7 @@ public void accept(RSocket rSocket, Throwable t) {
245244
return;
246245
}
247246

248-
startTime = ((DefaultPooledRSocket) parent).stats.startRequest();
247+
startTime = ((PooledWeightedRSocket) parent).stats.startRequest();
249248

250249
source.subscribe((CoreSubscriber) this);
251250
} else {
@@ -257,7 +256,7 @@ public void accept(RSocket rSocket, Throwable t) {
257256
public void onComplete() {
258257
final long state = this.requested;
259258
if (state != TERMINATED_STATE && REQUESTED.compareAndSet(this, state, TERMINATED_STATE)) {
260-
final Stats stats = ((DefaultPooledRSocket) parent).stats;
259+
final Stats stats = ((PooledWeightedRSocket) parent).stats;
261260
final long now = stats.stopRequest(startTime);
262261
stats.record(now - startTime);
263262
super.onComplete();
@@ -268,7 +267,7 @@ public void onComplete() {
268267
public void onError(Throwable t) {
269268
final long state = this.requested;
270269
if (state != TERMINATED_STATE && REQUESTED.compareAndSet(this, state, TERMINATED_STATE)) {
271-
Stats stats = ((DefaultPooledRSocket) parent).stats;
270+
Stats stats = ((PooledWeightedRSocket) parent).stats;
272271
stats.stopRequest(startTime);
273272
stats.recordError(0.0);
274273
super.onError(t);
@@ -284,7 +283,7 @@ public void cancel() {
284283

285284
if (state == STATE_SUBSCRIBED) {
286285
this.s.cancel();
287-
((DefaultPooledRSocket) parent).stats.stopRequest(startTime);
286+
((PooledWeightedRSocket) parent).stats.stopRequest(startTime);
288287
} else {
289288
this.parent.remove(this);
290289
ReferenceCountUtil.safeRelease(this.payload);
@@ -296,7 +295,7 @@ static final class RequestTrackingFluxInner<INPUT>
296295
extends FluxDeferredResolution<INPUT, RSocket> {
297296

298297
RequestTrackingFluxInner(
299-
DefaultPooledRSocket parent, INPUT fluxOrPayload, FrameType requestType) {
298+
PooledWeightedRSocket parent, INPUT fluxOrPayload, FrameType requestType) {
300299
super(parent, fluxOrPayload, requestType);
301300
}
302301

@@ -329,7 +328,7 @@ public void accept(RSocket rSocket, Throwable t) {
329328
return;
330329
}
331330

332-
((DefaultPooledRSocket) parent).stats.startStream();
331+
((PooledWeightedRSocket) parent).stats.startStream();
333332

334333
source.subscribe(this);
335334
} else {
@@ -341,7 +340,7 @@ public void accept(RSocket rSocket, Throwable t) {
341340
public void onComplete() {
342341
final long state = this.requested;
343342
if (state != TERMINATED_STATE && REQUESTED.compareAndSet(this, state, TERMINATED_STATE)) {
344-
((DefaultPooledRSocket) parent).stats.stopStream();
343+
((PooledWeightedRSocket) parent).stats.stopStream();
345344
super.onComplete();
346345
}
347346
}
@@ -350,7 +349,7 @@ public void onComplete() {
350349
public void onError(Throwable t) {
351350
final long state = this.requested;
352351
if (state != TERMINATED_STATE && REQUESTED.compareAndSet(this, state, TERMINATED_STATE)) {
353-
((DefaultPooledRSocket) parent).stats.stopStream();
352+
((PooledWeightedRSocket) parent).stats.stopStream();
354353
super.onError(t);
355354
}
356355
}
@@ -364,7 +363,7 @@ public void cancel() {
364363

365364
if (state == STATE_SUBSCRIBED) {
366365
this.s.cancel();
367-
((DefaultPooledRSocket) parent).stats.stopStream();
366+
((PooledWeightedRSocket) parent).stats.stopStream();
368367
} else {
369368
this.parent.remove(this);
370369
if (requestType == FrameType.REQUEST_STREAM) {

‎rsocket-core/src/main/java/io/rsocket/loadbalance/RSocketPool.java

Lines changed: 25 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -37,20 +37,20 @@
3737
import reactor.util.annotation.Nullable;
3838

3939
class RSocketPool extends ResolvingOperator<Void>
40-
implements CoreSubscriber<List<LoadbalanceRSocketSource>>, List<PooledRSocket> {
40+
implements CoreSubscriber<List<LoadbalanceRSocketSource>>, List<WeightedRSocket> {
4141

4242
final DeferredResolutionRSocket deferredResolutionRSocket = new DeferredResolutionRSocket(this);
4343
final LoadbalanceStrategy loadbalanceStrategy;
4444
final Supplier<Stats> statsSupplier;
4545

46-
volatile PooledRSocket[] activeSockets;
46+
volatile PooledWeightedRSocket[] activeSockets;
4747

48-
static final AtomicReferenceFieldUpdater<RSocketPool, PooledRSocket[]> ACTIVE_SOCKETS =
48+
static final AtomicReferenceFieldUpdater<RSocketPool, PooledWeightedRSocket[]> ACTIVE_SOCKETS =
4949
AtomicReferenceFieldUpdater.newUpdater(
50-
RSocketPool.class, PooledRSocket[].class, "activeSockets");
50+
RSocketPool.class, PooledWeightedRSocket[].class, "activeSockets");
5151

52-
static final PooledRSocket[] EMPTY = new PooledRSocket[0];
53-
static final PooledRSocket[] TERMINATED = new PooledRSocket[0];
52+
static final PooledWeightedRSocket[] EMPTY = new PooledWeightedRSocket[0];
53+
static final PooledWeightedRSocket[] TERMINATED = new PooledWeightedRSocket[0];
5454

5555
volatile Subscription s;
5656
static final AtomicReferenceFieldUpdater<RSocketPool, Subscription> S =
@@ -96,8 +96,8 @@ public void onNext(List<LoadbalanceRSocketSource> loadbalanceRSocketSources) {
9696
return;
9797
}
9898

99-
PooledRSocket[] previouslyActiveSockets;
100-
PooledRSocket[] activeSockets;
99+
PooledWeightedRSocket[] previouslyActiveSockets;
100+
PooledWeightedRSocket[] activeSockets;
101101
for (; ; ) {
102102
HashMap<LoadbalanceRSocketSource, Integer> rSocketSuppliersCopy = new HashMap<>();
103103

@@ -108,11 +108,11 @@ public void onNext(List<LoadbalanceRSocketSource> loadbalanceRSocketSources) {
108108

109109
// checking intersection of active RSocket with the newly received set
110110
previouslyActiveSockets = this.activeSockets;
111-
PooledRSocket[] nextActiveSockets =
112-
new PooledRSocket[previouslyActiveSockets.length + rSocketSuppliersCopy.size()];
111+
PooledWeightedRSocket[] nextActiveSockets =
112+
new PooledWeightedRSocket[previouslyActiveSockets.length + rSocketSuppliersCopy.size()];
113113
int position = 0;
114114
for (int i = 0; i < previouslyActiveSockets.length; i++) {
115-
PooledRSocket rSocket = previouslyActiveSockets[i];
115+
PooledWeightedRSocket rSocket = previouslyActiveSockets[i];
116116

117117
Integer index = rSocketSuppliersCopy.remove(rSocket.source());
118118
if (index == null) {
@@ -130,7 +130,7 @@ public void onNext(List<LoadbalanceRSocketSource> loadbalanceRSocketSources) {
130130
} else {
131131
// put newly create RSocket instance
132132
nextActiveSockets[position++] =
133-
new DefaultPooledRSocket(
133+
new PooledWeightedRSocket(
134134
this, loadbalanceRSocketSources.get(index), this.statsSupplier.get());
135135
}
136136
}
@@ -139,7 +139,7 @@ public void onNext(List<LoadbalanceRSocketSource> loadbalanceRSocketSources) {
139139
// going though brightly new rsocket
140140
for (LoadbalanceRSocketSource newLoadbalanceRSocketSource : rSocketSuppliersCopy.keySet()) {
141141
nextActiveSockets[position++] =
142-
new DefaultPooledRSocket(this, newLoadbalanceRSocketSource, this.statsSupplier.get());
142+
new PooledWeightedRSocket(this, newLoadbalanceRSocketSource, this.statsSupplier.get());
143143
}
144144

145145
// shrank to actual length
@@ -198,7 +198,7 @@ RSocket select() {
198198

199199
@Nullable
200200
RSocket doSelect() {
201-
PooledRSocket[] sockets = this.activeSockets;
201+
WeightedRSocket[] sockets = this.activeSockets;
202202
if (sockets == EMPTY) {
203203
return null;
204204
}
@@ -207,7 +207,7 @@ RSocket doSelect() {
207207
}
208208

209209
@Override
210-
public PooledRSocket get(int index) {
210+
public WeightedRSocket get(int index) {
211211
return activeSockets[index];
212212
}
213213

@@ -361,12 +361,12 @@ public boolean contains(Object o) {
361361
}
362362

363363
@Override
364-
public Iterator<PooledRSocket> iterator() {
364+
public Iterator<WeightedRSocket> iterator() {
365365
throw new UnsupportedOperationException();
366366
}
367367

368368
@Override
369-
public boolean add(PooledRSocket pooledRSocket) {
369+
public boolean add(WeightedRSocket weightedRSocket) {
370370
throw new UnsupportedOperationException();
371371
}
372372

@@ -381,12 +381,12 @@ public boolean containsAll(Collection<?> c) {
381381
}
382382

383383
@Override
384-
public boolean addAll(Collection<? extends PooledRSocket> c) {
384+
public boolean addAll(Collection<? extends WeightedRSocket> c) {
385385
throw new UnsupportedOperationException();
386386
}
387387

388388
@Override
389-
public boolean addAll(int index, Collection<? extends PooledRSocket> c) {
389+
public boolean addAll(int index, Collection<? extends WeightedRSocket> c) {
390390
throw new UnsupportedOperationException();
391391
}
392392

@@ -406,17 +406,17 @@ public void clear() {
406406
}
407407

408408
@Override
409-
public PooledRSocket set(int index, PooledRSocket element) {
409+
public WeightedRSocket set(int index, WeightedRSocket element) {
410410
throw new UnsupportedOperationException();
411411
}
412412

413413
@Override
414-
public void add(int index, PooledRSocket element) {
414+
public void add(int index, WeightedRSocket element) {
415415
throw new UnsupportedOperationException();
416416
}
417417

418418
@Override
419-
public PooledRSocket remove(int index) {
419+
public WeightedRSocket remove(int index) {
420420
throw new UnsupportedOperationException();
421421
}
422422

@@ -431,17 +431,17 @@ public int lastIndexOf(Object o) {
431431
}
432432

433433
@Override
434-
public ListIterator<PooledRSocket> listIterator() {
434+
public ListIterator<WeightedRSocket> listIterator() {
435435
throw new UnsupportedOperationException();
436436
}
437437

438438
@Override
439-
public ListIterator<PooledRSocket> listIterator(int index) {
439+
public ListIterator<WeightedRSocket> listIterator(int index) {
440440
throw new UnsupportedOperationException();
441441
}
442442

443443
@Override
444-
public List<PooledRSocket> subList(int fromIndex, int toIndex) {
444+
public List<WeightedRSocket> subList(int fromIndex, int toIndex) {
445445
throw new UnsupportedOperationException();
446446
}
447447
}

‎rsocket-core/src/main/java/io/rsocket/loadbalance/RoundRobinLoadbalanceStrategy.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ public class RoundRobinLoadbalanceStrategy implements LoadbalanceStrategy {
2626
AtomicIntegerFieldUpdater.newUpdater(RoundRobinLoadbalanceStrategy.class, "nextIndex");
2727

2828
@Override
29-
public PooledRSocket select(List<PooledRSocket> sockets) {
29+
public WeightedRSocket select(List<WeightedRSocket> sockets) {
3030
int length = sockets.size();
3131

3232
int indexToUse = Math.abs(NEXT_INDEX.getAndIncrement(this) % length);

0 commit comments

Comments
 (0)