Repository: hyperledger/quilt
Pull request #315 (2019-10-01 23:48:12)

fix bug with sendMoney between accounts in different assets
track how much money was sent from the sender's perspective instead of receiver's
fix issue with ScheduledThreadPoolExecutor not getting shutdown

Signed-off-by: nhartner <nhartner@gmail.com>

codecov[bot]: # [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=h1) Report
> Merging [#315](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=desc) into [master](https://codecov.io/gh/hyperledger/quilt/commit/8aeb61a1b7a48c830ce622f76797c0afc8c6332c?src=pr&el=desc) will **decrease** coverage by `0.08%`.
> The diff coverage is `90.47%`.

[![Impacted file tree graph](https://codecov.io/gh/hyperledger/quilt/pull/315/graphs/tree.svg?width=650&token=XEfCq6zC6W&height=150&src=pr)](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #315      +/-   ##
============================================
- Coverage     78.25%   78.17%   -0.09%     
  Complexity      754      754              
============================================
  Files           224      224              
  Lines          3486     3496      +10     
  Branches        222      222              
============================================
+ Hits           2728     2733       +5     
- Misses          664      668       +4     
- Partials         94       95       +1
```


| [Impacted Files](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...n/java/org/interledger/stream/SendMoneyResult.java](https://codecov.io/gh/hyperledger/quilt/pull/315/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vU2VuZE1vbmV5UmVzdWx0LmphdmE=) | `100% <100%> (ø)` | `3 <1> (ø)` | :arrow_down: |
| [.../interledger/stream/sender/SimpleStreamSender.java](https://codecov.io/gh/hyperledger/quilt/pull/315/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL3N0cmVhbS9zZW5kZXIvU2ltcGxlU3RyZWFtU2VuZGVyLmphdmE=) | `62.59% <90%> (-0.52%)` | `0 <0> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=footer). Last update [8aeb61a...bf3eab2](https://codecov.io/gh/hyperledger/quilt/pull/315?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/examples/send-money/pom.xml b/examples/send-money/pom.xml
index e81efdf..aa4a48c 100644
--- a/examples/send-money/pom.xml
+++ b/examples/send-money/pom.xml
@@ -62,6 +62,11 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-jdk14</artifactId>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 
 </project>
diff --git a/examples/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java b/examples/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
index dbe3c1b..3869ea1 100644
--- a/examples/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
+++ b/examples/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
@@ -15,6 +15,7 @@ import org.interledger.spsp.client.rust.InterledgerRustNodeClient;
 import org.interledger.stream.SendMoneyResult;
 import org.interledger.core.SharedSecret;
 import org.interledger.stream.sender.SimpleStreamSender;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -35,12 +36,10 @@ public class SendMoneyExample {
   // NOTE - replace this with the passkey for your sender account
   private static final String SENDER_PASS_KEY = "jxelaxvqz2ne6";
   // NOTE - replace this with the payment pointer for your receiver account
-  private static final String RECEIVER_PAYMENT_POINTER = "$rs3.xpring.dev/accounts/user_jwmqq7kv/spsp";
+  private static final String RECEIVER_PAYMENT_POINTER = "$rs3.xpring.dev/accounts/user_uk152ktl/spsp";
 
   private static final String TESTNET_URI = "https://rs3.xpring.dev";
 
-  private static final int ASSET_SCALE = 9;
-
   private static final InterledgerAddress SENDER_ADDRESS =
       InterledgerAddress.of("test.xpring-dev.rs3").with(SENDER_ACCOUNT_USERNAME);
 
@@ -63,7 +62,7 @@ public class SendMoneyExample {
 
     // Send payment using STREAM
     SendMoneyResult result = simpleStreamSender.sendMoney(SharedSecret.of(connectionDetails.sharedSecret().value()),
-        SENDER_ADDRESS, connectionDetails.destinationAddress(), UnsignedLong.valueOf(1000))
+        SENDER_ADDRESS, connectionDetails.destinationAddress(), UnsignedLong.valueOf(100000), Duration.ofMillis(30000))
         .get();
 
     System.out.println("Send money result: " + result);
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
index d7b8de6..602cd2f 100644
--- a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
@@ -56,8 +56,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -107,16 +106,15 @@ public class SimpleStreamSender implements StreamSender {
    */
   public SimpleStreamSender(final StreamEncryptionService streamEncryptionService, final Link link) {
     this(
-        streamEncryptionService, link,
-        new ThreadPoolExecutor(
-            0, 30, 60L, TimeUnit.SECONDS,
-            new SynchronousQueue<>(),
-            new ThreadFactoryBuilder()
-                .setDaemon(true)
-                .setNameFormat("simple-stream-sender-%d")
-                .build()
-        )
-    );
+        streamEncryptionService, link, newDefaultExecutor());
+  }
+
+  public static ExecutorService newDefaultExecutor() {
+    ThreadFactory factory = new ThreadFactoryBuilder()
+        .setDaemon(true)
+        .setNameFormat("simple-stream-sender-%d")
+        .build();
+    return Executors.newFixedThreadPool(30, factory);
   }
 
 
@@ -286,6 +284,7 @@ public class SimpleStreamSender implements StreamSender {
     private AtomicReference<UnsignedLong> originalAmountToSend;
     private AtomicReference<UnsignedLong> amountLeftToSend;
     private AtomicReference<UnsignedLong> deliveredAmount;
+    private final AtomicReference<UnsignedLong> sentAmount;
     private AtomicBoolean shouldSendSourceAddress;
     private AtomicInteger numFulfilledPackets;
     private AtomicInteger numRejectedPackets;
@@ -335,6 +334,7 @@ public class SimpleStreamSender implements StreamSender {
 
       this.originalAmountToSend = new AtomicReference<>(originalAmountToSend);
       this.amountLeftToSend = new AtomicReference<>(originalAmountToSend);
+      this.sentAmount = new AtomicReference<>(UnsignedLong.ZERO);
       this.deliveredAmount = new AtomicReference<>(UnsignedLong.ZERO);
 
       this.numFulfilledPackets = new AtomicInteger(0);
@@ -360,9 +360,10 @@ public class SimpleStreamSender implements StreamSender {
 
             // Do all the work of sending packetized money for this Stream/sendMoney request.
             this.sendMoneyPacketized();
-
             return SendMoneyResult.builder()
                 .amountDelivered(deliveredAmount.get())
+                .amountSent(sentAmount.get())
+                .amountLeftToSend(this.amountLeftToSend.get())
                 .originalAmount(originalAmountToSend.get())
                 .numFulfilledPackets(numFulfilledPackets.get())
                 .numRejectPackets(numRejectedPackets.get())
@@ -373,7 +374,7 @@ public class SimpleStreamSender implements StreamSender {
             if (error != null) {
               logger.error("SendMoney Stream failed: " + error.getMessage(), error);
             }
-            if (!originalAmountToSend.get().equals(this.deliveredAmount.get())) {
+            if (!$.successfulPayment()) {
               logger.error("Failed to send full amount");
             }
           });
@@ -387,8 +388,9 @@ public class SimpleStreamSender implements StreamSender {
 
       final AtomicBoolean timeoutReached = new AtomicBoolean(false);
 
+      final ScheduledExecutorService timeoutMonitor = Executors.newSingleThreadScheduledExecutor();
+
       timeout.ifPresent($ -> {
-        ScheduledExecutorService timeoutMonitor = Executors.newSingleThreadScheduledExecutor();
         timeoutMonitor.schedule(
             () -> {
               timeoutReached.set(true);
@@ -412,8 +414,6 @@ public class SimpleStreamSender implements StreamSender {
           continue;
         }
 
-        this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.minus(amountToSend));
-
         // Load up the STREAM packet
         final UnsignedLong sequence;
         try {
@@ -467,6 +467,8 @@ public class SimpleStreamSender implements StreamSender {
             .data(streamPacketData)
             .build();
 
+        this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.minus(amountToSend));
+
         try {
           // don't submit new tasks if the timeout was reached within this iteration of the while loop
           if (!timeoutReached.get()) {
@@ -518,6 +520,7 @@ public class SimpleStreamSender implements StreamSender {
           logger.error("Submit failed", e);
         }
       }
+      timeoutMonitor.shutdownNow();
     }
 
     @VisibleForTesting
@@ -604,6 +607,7 @@ public class SimpleStreamSender implements StreamSender {
       if (streamPacket.interledgerPacketType() == InterledgerPacketType.FULFILL) {
         // TODO check that the sequence matches our outgoing packet
         this.deliveredAmount.getAndUpdate(currentAmount -> currentAmount.plus(streamPacket.prepareAmount()));
+        this.sentAmount.getAndUpdate(currentAmount -> currentAmount.plus(UnsignedLong.valueOf(originalPreparePacket.getAmount())));
       } else {
         logger.warn("Unable to parse STREAM packet from fulfill data. "
                 + "originalPreparePacket={} originalStreamPacket={} fulfillPacket={}",
@@ -770,6 +774,8 @@ public class SimpleStreamSender implements StreamSender {
     SendMoneyResult collectSendMoneyStatistics() {
       return SendMoneyResult.builder()
           .amountDelivered(this.deliveredAmount.get())
+          .amountSent(this.sentAmount.get())
+          .amountLeftToSend(this.amountLeftToSend.get())
           .numFulfilledPackets(this.numFulfilledPackets.get())
           .numRejectPackets(this.numRejectedPackets.get())
           .build();
@@ -777,18 +783,18 @@ public class SimpleStreamSender implements StreamSender {
 
     @VisibleForTesting
     boolean moreToSend() {
-      return this.deliveredAmount.get().compareTo(this.originalAmountToSend.get()) < 0;
+      return this.sentAmount.get().compareTo(this.originalAmountToSend.get()) < 0;
     }
 
     /**
      * Helper method to mess with delivered amount for the purpose of testing loop breaking conditions.
      *
-     * @param deliveredAmount An {@link UnsignedLong} representing the deliveredAmount to set.
+     * @param sentAmount An {@link UnsignedLong} representing the deliveredAmount to set.
      */
     @VisibleForTesting
-    void setDeliveredAmountForTesting(final UnsignedLong deliveredAmount) {
-      Objects.requireNonNull(deliveredAmount);
-      this.deliveredAmount.set(deliveredAmount);
+    void setAmountSentForTesting(final UnsignedLong sentAmount) {
+      Objects.requireNonNull(sentAmount);
+      this.sentAmount.set(sentAmount);
     }
   }
 }
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyResult.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyResult.java
index 3ae332c..d1e1f19 100644
--- a/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyResult.java
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyResult.java
@@ -25,12 +25,30 @@ public interface SendMoneyResult {
   UnsignedLong originalAmount();
 
   /**
-   * The actual amount, in the senders units, that was delivered to the receiver.
+   * The actual amount, in the receivers units, that was delivered to the receiver. Any currency conversion and/or
+   * connector fees may cause this to be different than the amount sent.
    *
    * @return An {@link UnsignedLong} representing the amount delivered.
    */
   UnsignedLong amountDelivered();
 
+  /**
+   * The actual amount, in the senders units, that was sent to the receiver. In the case, of a timeout or rejected
+   * packets this amount may be less than the requested amount to be sent.
+   *
+   * @return An {@link UnsignedLong} representing the amount sent.
+   */
+  UnsignedLong amountSent();
+
+  /**
+   * The actual amount, in the senders units, that is still left to send. If the payment was successful, this amount
+   * will be 0. If there was an issue sending payment (repeated rejections or timeouts), this will be the amount that
+   * could not be sent or which may have been sent but we failed to receive the fulfillment packet (network issue).
+   *
+   * @return An {@link UnsignedLong} representing the amount left to send.
+   */
+  UnsignedLong amountLeftToSend();
+
   /**
    * Metadata representing the number of fulfilled packets in a request to send money. Note that STREAM implementations
    * MUST close the connection once either endpoint has sent 2^31 packets.
@@ -70,6 +88,6 @@ public interface SendMoneyResult {
    */
   @Value.Derived
   default boolean successfulPayment() {
-    return amountDelivered().equals(originalAmount());
+    return amountSent().equals(originalAmount());
   }
 }


Test patch:
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
index 1e67488..62ff84b 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
@@ -195,9 +195,9 @@ public class SendMoneyAggregatorTest {
     when(congestionControllerMock.hasInFlight()).thenReturn(moneyInFlight);
     when(streamConnectionMock.isClosed()).thenReturn(streamConnectionClosed);
     if (moreToSend) {
-      sendMoneyAggregator.setDeliveredAmountForTesting(UnsignedLong.ZERO);
+      sendMoneyAggregator.setAmountSentForTesting(UnsignedLong.ZERO);
     } else {
-      sendMoneyAggregator.setDeliveredAmountForTesting(UnsignedLong.valueOf(10L));
+      sendMoneyAggregator.setAmountSentForTesting(UnsignedLong.valueOf(10L));
     }
   }
 }
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/SendMoneyResultTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/SendMoneyResultTest.java
index ad93805..c3468a2 100644
--- a/stream-parent/stream-core/src/test/java/org/interledger/stream/SendMoneyResultTest.java
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/SendMoneyResultTest.java
@@ -17,6 +17,8 @@ public class SendMoneyResultTest {
     SendMoneyResult result = SendMoneyResult.builder()
         .originalAmount(UnsignedLong.ZERO)
         .amountDelivered(UnsignedLong.ZERO)
+        .amountSent(UnsignedLong.ZERO)
+        .amountLeftToSend(UnsignedLong.ZERO)
         .numFulfilledPackets(10)
         .numRejectPackets(5)
         .sendMoneyDuration(Duration.ZERO)