Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[streaming] ping related cleanup, inspection cleanup #3533

Merged
merged 5 commits into from
May 25, 2020
Merged

[streaming] ping related cleanup, inspection cleanup #3533

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
dd91b09
Merge pull request #1 from knowm/develop
earce May 21, 2020
7f93a8e
cleanup of streaming classes, removing inspection errors
earce May 23, 2020
1c4f174
Merge pull request #2 from knowm/develop
earce May 23, 2020
719b759
Merge remote-tracking branch 'origin/develop' into develop
earce May 23, 2020
8989b79
cleanup of streaming classes, removing inspection errors
earce May 24, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions ...gemini/src/main/java/info/bitrich/xchangestream/gemini/GeminiProductStreamingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@

import com.fasterxml.jackson.databind.JsonNode;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import java.io.IOException;
import org.knowm.xchange.currency.CurrencyPair;
import org.slf4j.Logger;
Expand All @@ -16,7 +14,12 @@ public class GeminiProductStreamingService extends JsonNettyStreamingService {
private final CurrencyPair currencyPair;

public GeminiProductStreamingService(String symbolUrl, CurrencyPair currencyPair) {
super(symbolUrl, Integer.MAX_VALUE, DEFAULT_CONNECTION_TIMEOUT, DEFAULT_RETRY_DURATION, 15);
super(
symbolUrl,
Integer.MAX_VALUE,
DEFAULT_CONNECTION_TIMEOUT,
DEFAULT_RETRY_DURATION,
DEFAULT_IDLE_TIMEOUT);
this.currencyPair = currencyPair;
}

Expand All @@ -39,9 +42,4 @@ public String getSubscribeMessage(String channelName, Object... args) throws IOE
public String getUnsubscribeMessage(String channelName) throws IOException {
return null;
}

@Override
protected void handleIdle(ChannelHandlerContext ctx) {
ctx.writeAndFlush(new PingWebSocketFrame());
}
}
24 changes: 4 additions & 20 deletions ...stream-kraken/src/main/java/info/bitrich/xchangestream/kraken/KrakenStreamingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,11 @@
import info.bitrich.xchangestream.service.netty.WebSocketClientHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.reactivex.Completable;
import io.reactivex.Observable;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -33,7 +30,7 @@ public class KrakenStreamingService extends JsonNettyStreamingService {
private static final Logger LOG = LoggerFactory.getLogger(KrakenStreamingService.class);
private static final String EVENT = "event";
private final Map<Integer, String> channels = new ConcurrentHashMap<>();
private ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
private final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
private final boolean isPrivate;

private final Map<Integer, String> subscriptionRequestMap = new ConcurrentHashMap<>();
Expand All @@ -57,6 +54,9 @@ protected void handleMessage(JsonNode message) {
KrakenEventType krakenEvent;
if (event != null && (krakenEvent = KrakenEventType.getEvent(event.textValue())) != null) {
switch (krakenEvent) {
case pingStatus:
LOG.info("PingStatus received: {}", message);
break;
case pong:
LOG.debug("Pong received");
break;
Expand Down Expand Up @@ -210,22 +210,6 @@ protected WebSocketClientHandler getWebSocketClientHandler(
return new KrakenWebSocketClientHandler(handshaker, handler);
}

@Override
protected Completable openConnection() {

KrakenSubscriptionMessage ping =
new KrakenSubscriptionMessage(null, KrakenEventType.ping, null, null);

subscribeConnectionSuccess()
.subscribe(
o ->
Observable.interval(30, TimeUnit.SECONDS)
.takeWhile(t -> isSocketOpen())
.subscribe(t -> sendObjectMessage(ping)));

return super.openConnection();
}

private WebSocketClientHandler.WebSocketMessageHandler channelInactiveHandler = null;

/**
Expand Down
1 change: 1 addition & 0 deletions ...eam-kraken/src/main/java/info/bitrich/xchangestream/kraken/dto/enums/KrakenEventType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ public enum KrakenEventType {
unsubscribe,
systemStatus,
subscriptionStatus,
pingStatus,
ping,
pong,
error;
Expand Down
7 changes: 0 additions & 7 deletions xchange-stream-lgo/src/main/java/info/bitrich/xchangestream/lgo/LgoStreamingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,7 @@
import com.fasterxml.jackson.databind.JsonNode;
import info.bitrich.xchangestream.lgo.dto.LgoSubscription;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import java.io.IOException;
import java.time.Duration;
import org.knowm.xchange.lgo.service.LgoSignatureService;
Expand All @@ -23,11 +21,6 @@ public class LgoStreamingService extends JsonNettyStreamingService {
this.signatureService = signatureService;
}

@Override
protected void handleIdle(ChannelHandlerContext ctx) {
ctx.writeAndFlush(new PingWebSocketFrame());
}

@Override
protected String getChannelNameFromMessage(JsonNode message) {
String channel = message.get("channel").asText();
Expand Down
7 changes: 2 additions & 5 deletions ...e-netty/src/main/java/info/bitrich/xchangestream/service/netty/NettyStreamingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,8 +163,7 @@ protected Completable openConnection() {
if (ssl) {
SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
if (acceptAllCertificates) {
sslContextBuilder =
sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like the fmt plugin needs rerunning here

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reran (nothing changed here but gemini streaming service did)

}
sslCtx = sslContextBuilder.build();
} else {
Expand Down Expand Up @@ -237,9 +236,7 @@ protected void initChannel(SocketChannel ch) {
webSocketChannel
.disconnect()
.addListener(
x -> {
completable.onError(handshakeFuture.cause());
});
x -> completable.onError(handshakeFuture.cause()));
}
});
} else {
Expand Down