Skip to content

Commit 47780c9

Browse files
steveguryNiteshKant
stevegury
authored and
NiteshKant
committed
Remove ReactiveSocketFactory.remote() (#145)
***Problem*** Since the refactoring of the `LoadBalancer` to use `List` instead of `Map`, there's no need for a `remote` method that identify a remote server. Furthermore, that's the only reason why LoadBalancer is generic. ***Solution*** Delete the method `ReactiveSocketFactory.remote()` as well as all the references to it.
1 parent 89e08fb commit 47780c9

9 files changed

Lines changed: 61 additions & 102 deletions

File tree

‎reactivesocket-client/src/main/java/io/reactivesocket/client/ClientBuilder.java‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -134,9 +134,9 @@ public void request(long n) {
134134
}
135135
filterConnector = filterConnector.chain(DrainingSocket::new);
136136

137-
Publisher<? extends Collection<ReactiveSocketFactory<T>>> factories =
137+
Publisher<? extends Collection<ReactiveSocketFactory>> factories =
138138
sourceToFactory(source, filterConnector);
139-
LoadBalancer<T> loadBalancer = new LoadBalancer<>(factories);
139+
LoadBalancer loadBalancer = new LoadBalancer(factories);
140140

141141
scheduledFuture = executor.scheduleAtFixedRate(() -> {
142142
if (loadBalancer.availability() > 0 && !cancelled.get()) {
@@ -161,13 +161,13 @@ public void cancel() {
161161
};
162162
}
163163

164-
private Publisher<? extends Collection<ReactiveSocketFactory<T>>> sourceToFactory(
164+
private Publisher<? extends Collection<ReactiveSocketFactory>> sourceToFactory(
165165
Publisher<? extends Collection<T>> source,
166166
ReactiveSocketConnector<T> connector
167167
) {
168168
return subscriber ->
169169
source.subscribe(new Subscriber<Collection<T>>() {
170-
private Map<T, ReactiveSocketFactory<T>> current;
170+
private Map<T, ReactiveSocketFactory> current;
171171

172172
@Override
173173
public void onSubscribe(Subscription s) {
@@ -177,23 +177,23 @@ public void onSubscribe(Subscription s) {
177177

178178
@Override
179179
public void onNext(Collection<T> socketAddresses) {
180-
Map<T, ReactiveSocketFactory<T>> next = new HashMap<>(socketAddresses.size());
180+
Map<T, ReactiveSocketFactory> next = new HashMap<>(socketAddresses.size());
181181
for (T sa: socketAddresses) {
182-
ReactiveSocketFactory<T> factory = current.get(sa);
182+
ReactiveSocketFactory factory = current.get(sa);
183183
if (factory == null) {
184-
ReactiveSocketFactory<T> newFactory = connector.toFactory(sa);
184+
ReactiveSocketFactory newFactory = connector.toFactory(sa);
185185
if (connectTimeout > 0) {
186-
newFactory = new TimeoutFactory<>(newFactory, connectTimeout, connectTimeoutUnit, executor);
186+
newFactory = new TimeoutFactory(newFactory, connectTimeout, connectTimeoutUnit, executor);
187187
}
188-
newFactory = new FailureAwareFactory<>(newFactory);
188+
newFactory = new FailureAwareFactory(newFactory);
189189
next.put(sa, newFactory);
190190
} else {
191191
next.put(sa, factory);
192192
}
193193
}
194194

195195
current = next;
196-
List<ReactiveSocketFactory<T>> factories = new ArrayList<>(current.values());
196+
List<ReactiveSocketFactory> factories = new ArrayList<>(current.values());
197197
subscriber.onNext(factories);
198198
}
199199

‎reactivesocket-client/src/main/java/io/reactivesocket/client/LoadBalancer.java‎

Lines changed: 23 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@
5050
* pool of children ReactiveSockets.
5151
* It estimates the load of each ReactiveSocket based on statistics collected.
5252
*/
53-
public class LoadBalancer<T> implements ReactiveSocket {
53+
public class LoadBalancer implements ReactiveSocket {
5454
public static final double DEFAULT_EXP_FACTOR = 4.0;
5555
public static final double DEFAULT_LOWER_QUANTILE = 0.2;
5656
public static final double DEFAULT_HIGHER_QUANTILE = 0.8;
@@ -78,7 +78,7 @@ public class LoadBalancer<T> implements ReactiveSocket {
7878

7979
private int pendingSockets;
8080
private final List<WeightedSocket> activeSockets;
81-
private final List<ReactiveSocketFactory<T>> activeFactories;
81+
private final List<ReactiveSocketFactory> activeFactories;
8282
private final FactoriesRefresher factoryRefresher;
8383

8484
private Ewma pendings;
@@ -109,7 +109,7 @@ public class LoadBalancer<T> implements ReactiveSocket {
109109
* ReactiveSocket is closed. (unit is millisecond)
110110
*/
111111
public LoadBalancer(
112-
Publisher<? extends Collection<ReactiveSocketFactory<T>>> factories,
112+
Publisher<? extends Collection<ReactiveSocketFactory>> factories,
113113
double expFactor,
114114
double lowQuantile,
115115
double highQuantile,
@@ -144,7 +144,7 @@ public LoadBalancer(
144144
factories.subscribe(factoryRefresher);
145145
}
146146

147-
public LoadBalancer(Publisher<? extends Collection<ReactiveSocketFactory<T>>> factories) {
147+
public LoadBalancer(Publisher<? extends Collection<ReactiveSocketFactory>> factories) {
148148
this(factories,
149149
DEFAULT_EXP_FACTOR,
150150
DEFAULT_LOWER_QUANTILE, DEFAULT_HIGHER_QUANTILE,
@@ -196,16 +196,16 @@ private synchronized void addSockets(int numberOfNewSocket) {
196196
while (n > 0) {
197197
int size = activeFactories.size();
198198
if (size == 1) {
199-
ReactiveSocketFactory<T> factory = activeFactories.get(0);
199+
ReactiveSocketFactory factory = activeFactories.get(0);
200200
if (factory.availability() > 0.0) {
201201
activeFactories.remove(0);
202202
pendingSockets++;
203203
factory.apply().subscribe(new SocketAdder(factory));
204204
}
205205
break;
206206
}
207-
ReactiveSocketFactory<T> factory0 = null;
208-
ReactiveSocketFactory<T> factory1 = null;
207+
ReactiveSocketFactory factory0 = null;
208+
ReactiveSocketFactory factory1 = null;
209209
int i0 = 0;
210210
int i1 = 0;
211211
for (int i = 0; i < EFFORT; i++) {
@@ -323,10 +323,6 @@ private synchronized void quickSlowestRS() {
323323
return;
324324
}
325325

326-
activeSockets.forEach(value -> {
327-
logger.info("> " + value);
328-
});
329-
330326
WeightedSocket slowest = null;
331327
double lowestAvailability = Double.MAX_VALUE;
332328
for (WeightedSocket socket: activeSockets) {
@@ -500,7 +496,7 @@ public Publisher<Void> onClose() {
500496
* This subscriber role is to subscribe to the list of server identifier, and update the
501497
* factory list.
502498
*/
503-
private class FactoriesRefresher implements Subscriber<Collection<ReactiveSocketFactory<T>>> {
499+
private class FactoriesRefresher implements Subscriber<Collection<ReactiveSocketFactory>> {
504500
private Subscription subscription;
505501

506502
@Override
@@ -510,21 +506,21 @@ public void onSubscribe(Subscription subscription) {
510506
}
511507

512508
@Override
513-
public void onNext(Collection<ReactiveSocketFactory<T>> newFactories) {
509+
public void onNext(Collection<ReactiveSocketFactory> newFactories) {
514510
synchronized (LoadBalancer.this) {
515511

516-
Set<ReactiveSocketFactory<T>> current =
512+
Set<ReactiveSocketFactory> current =
517513
new HashSet<>(activeFactories.size() + activeSockets.size());
518514
current.addAll(activeFactories);
519515
for (WeightedSocket socket: activeSockets) {
520-
ReactiveSocketFactory<T> factory = socket.getFactory();
516+
ReactiveSocketFactory factory = socket.getFactory();
521517
current.add(factory);
522518
}
523519

524-
Set<ReactiveSocketFactory<T>> removed = new HashSet<>(current);
520+
Set<ReactiveSocketFactory> removed = new HashSet<>(current);
525521
removed.removeAll(newFactories);
526522

527-
Set<ReactiveSocketFactory<T>> added = new HashSet<>(newFactories);
523+
Set<ReactiveSocketFactory> added = new HashSet<>(newFactories);
528524
added.removeAll(current);
529525

530526
boolean changed = false;
@@ -541,9 +537,9 @@ public void onNext(Collection<ReactiveSocketFactory<T>> newFactories) {
541537
}
542538
}
543539
}
544-
Iterator<ReactiveSocketFactory<T>> it1 = activeFactories.iterator();
540+
Iterator<ReactiveSocketFactory> it1 = activeFactories.iterator();
545541
while (it1.hasNext()) {
546-
ReactiveSocketFactory<T> factory = it1.next();
542+
ReactiveSocketFactory factory = it1.next();
547543
if (removed.contains(factory)) {
548544
it1.remove();
549545
changed = true;
@@ -554,7 +550,7 @@ public void onNext(Collection<ReactiveSocketFactory<T>> newFactories) {
554550

555551
if (changed && logger.isDebugEnabled()) {
556552
String msg = "\nUpdated active factories (size: " + activeFactories.size() + ")\n";
557-
for (ReactiveSocketFactory<T> f : activeFactories) {
553+
for (ReactiveSocketFactory f : activeFactories) {
558554
msg += " + " + f + "\n";
559555
}
560556
msg += "Active sockets:\n";
@@ -583,9 +579,9 @@ void close() {
583579
}
584580

585581
private class SocketAdder implements Subscriber<ReactiveSocket> {
586-
private final ReactiveSocketFactory<T> factory;
582+
private final ReactiveSocketFactory factory;
587583

588-
private SocketAdder(ReactiveSocketFactory<T> factory) {
584+
private SocketAdder(ReactiveSocketFactory factory) {
589585
this.factory = factory;
590586
}
591587

@@ -602,8 +598,7 @@ public void onNext(ReactiveSocket rs) {
602598
}
603599

604600
WeightedSocket weightedSocket = new WeightedSocket(rs, factory, lowerQuantile, higherQuantile);
605-
logger.info("Adding new WeightedSocket "
606-
+ weightedSocket + " connected to " + factory.remote());
601+
logger.info("Adding new WeightedSocket {}", weightedSocket);
607602

608603
activeSockets.add(weightedSocket);
609604
pendingSockets -= 1;
@@ -711,7 +706,7 @@ private class WeightedSocket extends ReactiveSocketProxy {
711706
private static final double STARTUP_PENALTY = Long.MAX_VALUE >> 12;
712707

713708
private final ReactiveSocket child;
714-
private ReactiveSocketFactory<T> factory;
709+
private ReactiveSocketFactory factory;
715710
private final Quantile lowerQuantile;
716711
private final Quantile higherQuantile;
717712
private final long inactivityFactor;
@@ -728,7 +723,7 @@ private class WeightedSocket extends ReactiveSocketProxy {
728723

729724
WeightedSocket(
730725
ReactiveSocket child,
731-
ReactiveSocketFactory<T> factory,
726+
ReactiveSocketFactory factory,
732727
Quantile lowerQuantile,
733728
Quantile higherQuantile,
734729
int inactivityFactor
@@ -751,7 +746,7 @@ private class WeightedSocket extends ReactiveSocketProxy {
751746

752747
WeightedSocket(
753748
ReactiveSocket child,
754-
ReactiveSocketFactory<T> factory,
749+
ReactiveSocketFactory factory,
755750
Quantile lowerQuantile,
756751
Quantile higherQuantile
757752
) {
@@ -794,7 +789,7 @@ public Publisher<Payload> requestChannel(Publisher<Payload> payloads) {
794789
child.requestChannel(payloads).subscribe(new CountingSubscriber<>(subscriber, this));
795790
}
796791

797-
ReactiveSocketFactory<T> getFactory() {
792+
ReactiveSocketFactory getFactory() {
798793
return factory;
799794
}
800795

‎reactivesocket-client/src/main/java/io/reactivesocket/client/filter/FailureAwareFactory.java‎

Lines changed: 4 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
import io.reactivesocket.Payload;
1919
import io.reactivesocket.ReactiveSocket;
2020
import io.reactivesocket.ReactiveSocketFactory;
21-
import io.reactivesocket.exceptions.TransportException;
2221
import io.reactivesocket.client.util.Clock;
2322
import io.reactivesocket.client.stat.Ewma;
2423
import io.reactivesocket.util.ReactiveSocketProxy;
@@ -27,7 +26,6 @@
2726
import org.reactivestreams.Subscription;
2827

2928
import java.util.concurrent.TimeUnit;
30-
import java.util.function.Function;
3129

3230
/**
3331
* This child compute the error rate of a particular remote location and adapt the availability
@@ -36,25 +34,23 @@
3634
* It means that if a remote host doesn't generate lots of errors when connecting to it, but a
3735
* lot of them when sending messages, we will still decrease the availability of the child
3836
* reducing the probability of connecting to it.
39-
*
40-
* @param <T> the identifier for the remote server (most likely SocketAddress)
4137
*/
42-
public class FailureAwareFactory<T> implements ReactiveSocketFactory<T> {
38+
public class FailureAwareFactory implements ReactiveSocketFactory {
4339
private static final double EPSILON = 1e-4;
4440

45-
private final ReactiveSocketFactory<T> child;
41+
private final ReactiveSocketFactory child;
4642
private final long tau;
4743
private long stamp;
4844
private Ewma errorPercentage;
4945

50-
public FailureAwareFactory(ReactiveSocketFactory<T> child, long halfLife, TimeUnit unit) {
46+
public FailureAwareFactory(ReactiveSocketFactory child, long halfLife, TimeUnit unit) {
5147
this.child = child;
5248
this.tau = Clock.unit().convert((long)(halfLife / Math.log(2)), unit);
5349
this.stamp = Clock.now();
5450
errorPercentage = new Ewma(halfLife, unit, 1.0);
5551
}
5652

57-
public FailureAwareFactory(ReactiveSocketFactory<T> child) {
53+
public FailureAwareFactory(ReactiveSocketFactory child) {
5854
this(child, 5, TimeUnit.SECONDS);
5955
}
6056

@@ -102,11 +98,6 @@ public double availability() {
10298
return e;
10399
}
104100

105-
@Override
106-
public T remote() {
107-
return child.remote();
108-
}
109-
110101
private synchronized void updateErrorPercentage(double value) {
111102
errorPercentage.insert(value);
112103
stamp = Clock.now();

‎reactivesocket-client/src/main/java/io/reactivesocket/client/filter/TimeoutFactory.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,12 @@
2424
import java.util.concurrent.ScheduledExecutorService;
2525
import java.util.concurrent.TimeUnit;
2626

27-
public class TimeoutFactory<T> extends ReactiveSocketFactoryProxy<T> {
27+
public class TimeoutFactory extends ReactiveSocketFactoryProxy {
2828
private final Publisher<Void> timer;
2929
private final long timeout;
3030
private final TimeUnit unit;
3131

32-
public TimeoutFactory(ReactiveSocketFactory<T> child, long timeout, TimeUnit unit,
32+
public TimeoutFactory(ReactiveSocketFactory child, long timeout, TimeUnit unit,
3333
ScheduledExecutorService executor) {
3434
super(child);
3535
this.timeout = timeout;

‎reactivesocket-client/src/test/java/io/reactivesocket/client/FailureReactiveSocketTest.java‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ private void testReactiveSocket(BiConsumer<CountDownLatch, ReactiveSocket> f) th
115115
throw new RuntimeException();
116116
}
117117
});
118-
ReactiveSocketFactory<String> factory = new ReactiveSocketFactory<String>() {
118+
ReactiveSocketFactory factory = new ReactiveSocketFactory() {
119119
@Override
120120
public Publisher<ReactiveSocket> apply() {
121121
return subscriber -> {
@@ -129,13 +129,9 @@ public double availability() {
129129
return 1.0;
130130
}
131131

132-
@Override
133-
public String remote() {
134-
return "Testing";
135-
}
136132
};
137133

138-
FailureAwareFactory<String> failureFactory = new FailureAwareFactory<>(factory, 100, TimeUnit.MILLISECONDS);
134+
FailureAwareFactory failureFactory = new FailureAwareFactory(factory, 100, TimeUnit.MILLISECONDS);
139135

140136
CountDownLatch latch = new CountDownLatch(1);
141137
failureFactory.apply().subscribe(new Subscriber<ReactiveSocket>() {

0 commit comments

Comments
 (0)