Merge pull request #4959 from bigscoop/gateio-stream-fixes

[gateio] Streaming fixes

xchange-stream-gateio/.gitignore

@@ -1 +1,2 @@
-http-client.private.env.json
+http-client.private.env.json
+integration-test.env.properties

xchange-stream-gateio/README.md

@@ -0,0 +1,21 @@
+## Using IntelliJ Idea HTTP client
+
+There are *.http files stored in `src/test/resources/rest` that can be used with IntelliJ Idea HTTP Client.
+
+Some requests need authorization, so the api credentials have to be stored in `http-client.private.env.json` in module's root. Sample content can be found in `example.http-client.private.env.json`
+
+> [!CAUTION]
+> Never commit your api credentials to the repository!
+
+
+[HTTP Client documentation](https://www.jetbrains.com/help/idea/http-client-in-product-code-editor.html)
+
+## Running integration tests that require API keys
+
+Integration tests that require API keys read them from environment variables. They can be defined in `integration-test.env.properties`. Sample content can be found in `example.integration-test.env.properties`.
+
+If no keys are provided the integration tests that need them are skipped.
+
+> [!CAUTION]
+> Never commit your api credentials to the repository!
+

xchange-stream-gateio/example.integration-test.env.properties

@@ -0,0 +1,2 @@
+apiKey=change_me
+secretKey=change_me

xchange-stream-gateio/pom.xml

@@ -44,4 +44,20 @@
 
     </dependencies>
 
+    <build>
+
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-failsafe-plugin</artifactId>
+                    <configuration>
+                        <systemPropertiesFile>integration-test.env.properties</systemPropertiesFile>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+
+    </build>
+
 </project>

xchange-stream-gateio/src/main/java/info/bitrich/xchangestream/gateio/GateioStreamingService.java

@@ -57,18 +57,22 @@ protected String getChannelNameFromMessage(GateioWsNotification message) {
   }
 
   @Override
-  public Observable<GateioWsNotification> subscribeChannel(String channelName, Object... args) {
+  public String getSubscriptionUniqueId(String channelName, Object... args) {
     final CurrencyPair currencyPair =
         (args.length > 0 && args[0] instanceof CurrencyPair) ? ((CurrencyPair) args[0]) : null;
 
-    String uniqueChannelName =
-        String.format("%s%s%s", channelName, Config.CHANNEL_NAME_DELIMITER, currencyPair);
+    return String.format("%s%s%s", channelName, Config.CHANNEL_NAME_DELIMITER, currencyPair);
+  }
+
+  @Override
+  public Observable<GateioWsNotification> subscribeChannel(String channelName, Object... args) {
+    String uniqueChannelName = getSubscriptionUniqueId(channelName, args);
 
     // Example channel name key: spot.order_book-BTC/USDT
     if (!channels.containsKey(uniqueChannelName) && !subscriptions.containsKey(uniqueChannelName)) {
 
       // subscribe
-      Observable<GateioWsNotification> observable = super.subscribeChannel(uniqueChannelName, args);
+      Observable<GateioWsNotification> observable = super.subscribeChannel(channelName, args);
 
       // cache channel subscribtion
       subscriptions.put(uniqueChannelName, observable);

xchange-stream-gateio/src/test/java/info/bitrich/xchangestream/gateio/GateioStreamingAccountServiceIntegration.java

@@ -1,27 +1,25 @@
 package info.bitrich.xchangestream.gateio;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.observers.TestObserver;
-import org.apache.commons.lang3.StringUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.knowm.xchange.currency.Currency;
 import org.knowm.xchange.dto.account.Balance;
 
-@Disabled(
-    "Needs authenticated exchange and real balance change. Set env vars GATEIO_API_KEY/GATEIO_API_SECRET")
+@Slf4j
 class GateioStreamingAccountServiceIntegration extends GateioStreamingExchangeIT {
 
-  @BeforeEach
-  void authConfigured() {
-    assumeTrue(
-        StringUtils.isNotEmpty(exchange.getExchangeSpecification().getApiKey()), "Needs auth");
-    assumeTrue(
-        StringUtils.isNotEmpty(exchange.getExchangeSpecification().getSecretKey()), "Needs auth");
+  @BeforeAll
+  public static void credentialsPresent() {
+    // skip if there are no credentials
+    assumeThat(exchange.getExchangeSpecification().getApiKey()).isNotEmpty();
+    assumeThat(exchange.getExchangeSpecification().getSecretKey()).isNotEmpty();
   }
 
   @Test
@@ -31,10 +29,17 @@ void spot_balances() {
 
     TestObserver<Balance> testObserver = observable.test();
 
-    Balance balance = testObserver.awaitCount(1).values().get(0);
+    List<Balance> balances = testObserver
+//        .awaitDone(10, TimeUnit.MINUTES)
+        .awaitCount(1)
+        .values();
 
     testObserver.dispose();
 
-    assertThat(balance).hasNoNullFieldsOrProperties();
+    log.info("Received balances: {}", balances);
+
+    assumeThat(balances).overridingErrorMessage("Received nothing").isNotEmpty();
+
+    assertThat(balances).first().hasNoNullFieldsOrProperties();
   }
 }

xchange-stream-gateio/src/test/java/info/bitrich/xchangestream/gateio/GateioStreamingExchangeIT.java

@@ -20,8 +20,8 @@ public static void setup() {
           StreamingExchangeFactory.INSTANCE
               .createExchangeWithoutSpecification(GateioStreamingExchange.class)
               .getDefaultExchangeSpecification();
-      spec.setApiKey(System.getenv("GATEIO_API_KEY"));
-      spec.setSecretKey(System.getenv("GATEIO_API_SECRET"));
+      spec.setApiKey(System.getProperty("apiKey"));
+      spec.setSecretKey(System.getProperty("secretKey"));
 
       exchange = (GateioStreamingExchange) StreamingExchangeFactory.INSTANCE.createExchange(spec);
 

xchange-stream-gateio/src/test/java/info/bitrich/xchangestream/gateio/GateioStreamingMarketDataServiceIntegration.java

@@ -22,7 +22,10 @@ void order_book() {
 
     TestObserver<OrderBook> testObserver = observable.test();
 
-    OrderBook orderBook = testObserver.awaitCount(1).values().get(0);
+    OrderBook orderBook = testObserver
+//        .awaitDone(1, TimeUnit.MINUTES)
+        .awaitCount(1)
+        .values().get(0);
 
     testObserver.dispose();
 
@@ -42,7 +45,10 @@ void trades() {
 
     TestObserver<Trade> testObserver = observable.test();
 
-    Trade trade = testObserver.awaitCount(1).values().get(0);
+    Trade trade = testObserver
+//        .awaitDone(1, TimeUnit.MINUTES)
+        .awaitCount(1)
+        .values().get(0);
 
     testObserver.dispose();
 
@@ -57,7 +63,10 @@ void ticker() {
 
     TestObserver<Ticker> testObserver = observable.test();
 
-    Ticker ticker = testObserver.awaitCount(1).values().get(0);
+    Ticker ticker = testObserver
+        .awaitCount(1)
+//        .awaitDone(1, TimeUnit.MINUTES)
+        .values().get(0);
 
     testObserver.dispose();
 

xchange-stream-gateio/src/test/java/info/bitrich/xchangestream/gateio/GateioStreamingTradeServiceIntegration.java

@@ -1,54 +1,69 @@
 package info.bitrich.xchangestream.gateio;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import static org.assertj.core.api.Assumptions.assumeThat;
+import static org.knowm.xchange.currency.CurrencyPair.BTC_USDT;
 
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.observers.TestObserver;
-import org.apache.commons.lang3.StringUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
-import org.knowm.xchange.currency.CurrencyPair;
 import org.knowm.xchange.dto.trade.UserTrade;
 
-@Disabled(
-    "Needs authenticated exchange and real user trade. Set env vars GATEIO_API_KEY/GATEIO_API_SECRET")
+@Slf4j
 class GateioStreamingTradeServiceIntegration extends GateioStreamingExchangeIT {
 
-  @BeforeEach
-  void authConfigured() {
-    assumeTrue(
-        StringUtils.isNotEmpty(exchange.getExchangeSpecification().getApiKey()), "Needs auth");
-    assumeTrue(
-        StringUtils.isNotEmpty(exchange.getExchangeSpecification().getSecretKey()), "Needs auth");
+  @BeforeAll
+  public static void credentialsPresent() {
+    // skip if there are no credentials
+    assumeThat(exchange.getExchangeSpecification().getApiKey()).isNotEmpty();
+    assumeThat(exchange.getExchangeSpecification().getSecretKey()).isNotEmpty();
   }
 
   @Test
-  void user_trades_btc() {
-    Observable<UserTrade> observable =
-        exchange.getStreamingTradeService().getUserTrades(CurrencyPair.BTC_USDT);
+  void user_trades_all() {
+    Observable<UserTrade> observable = exchange.getStreamingTradeService().getUserTrades();
 
     TestObserver<UserTrade> testObserver = observable.test();
 
-    UserTrade userTrade = testObserver.awaitCount(1).values().get(0);
+    List<UserTrade> userTrades = testObserver
+//        .awaitDone(10, TimeUnit.MINUTES)
+        .awaitCount(1)
+        .values();
 
     testObserver.dispose();
 
-    assertThat(userTrade).hasNoNullFieldsOrPropertiesExcept("makerOrderId", "takerOrderId");
-    assertThat(userTrade.getInstrument()).isEqualTo(CurrencyPair.BTC_USDT);
+    log.info("Received usertrades: {}", userTrades);
+
+    assumeThat(userTrades).overridingErrorMessage("No trades happened").isNotEmpty();
+
+    assertThat(userTrades.get(0).getInstrument()).isNotNull();
+    assertThat(userTrades.get(0).getId()).isNotNull();
+    assertThat(userTrades.get(0).getOrderId()).isNotNull();
   }
 
   @Test
-  void user_trades_all() {
-    Observable<UserTrade> observable = exchange.getStreamingTradeService().getUserTrades();
+  void user_trades_single_instrument() {
+    Observable<UserTrade> observable = exchange.getStreamingTradeService().getUserTrades(BTC_USDT);
 
     TestObserver<UserTrade> testObserver = observable.test();
 
-    UserTrade userTrade = testObserver.awaitCount(1).values().get(0);
+    List<UserTrade> userTrades =
+        testObserver
+            //        .awaitDone(1, TimeUnit.MINUTES)
+            .awaitCount(1)
+            .values();
 
     testObserver.dispose();
 
-    assertThat(userTrade).hasNoNullFieldsOrPropertiesExcept("makerOrderId", "takerOrderId");
+    log.info("Received usertrades: {}", userTrades);
+
+    assumeThat(userTrades).overridingErrorMessage("No trades happened").isNotEmpty();
+
+    assertThat(userTrades.get(0).getInstrument()).isEqualTo(BTC_USDT);
+    assertThat(userTrades.get(0).getId()).isNotNull();
+    assertThat(userTrades.get(0).getOrderId()).isNotNull();
   }
 }