Repository: hyperledger/quilt
Pull request #318 (2019-10-03 20:31:29)

Is/exchange rates the sequel
Addresses https://github.com/hyperledger/quilt/issues/291

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

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

```diff
@@             Coverage Diff             @@
##             master    #318      +/-   ##
===========================================
- Coverage     78.97%   78.8%   -0.18%     
  Complexity      766     766              
===========================================
  Files           224     229       +5     
  Lines          3496    3581      +85     
  Branches        222     223       +1     
===========================================
+ Hits           2761    2822      +61     
- Misses          642     662      +20     
- Partials         93      97       +4
```


| [Impacted Files](https://codecov.io/gh/hyperledger/quilt/pull/318?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...ger/stream/calculators/ExchangeRateCalculator.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vY2FsY3VsYXRvcnMvRXhjaGFuZ2VSYXRlQ2FsY3VsYXRvci5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...er/stream/calculators/NoExchangeRateException.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vY2FsY3VsYXRvcnMvTm9FeGNoYW5nZVJhdGVFeGNlcHRpb24uamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...ain/java/org/interledger/stream/Denominations.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vRGVub21pbmF0aW9ucy5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...stream/calculators/NoOpExchangeRateCalculator.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vY2FsY3VsYXRvcnMvTm9PcEV4Y2hhbmdlUmF0ZUNhbGN1bGF0b3IuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [.../java/org/interledger/stream/SendMoneyRequest.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vU2VuZE1vbmV5UmVxdWVzdC5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [.../interledger/stream/sender/SimpleStreamSender.java](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL3N0cmVhbS9zZW5kZXIvU2ltcGxlU3RyZWFtU2VuZGVyLmphdmE=) | `71.47% <86.41%> (+7.69%)` | `0 <0> (ø)` | :arrow_down: |
| ... and [2 more](https://codecov.io/gh/hyperledger/quilt/pull/318/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hyperledger/quilt/pull/318?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/318?src=pr&el=footer). Last update [477d1ed...9623047](https://codecov.io/gh/hyperledger/quilt/pull/318?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

theotherian: Also addresses #306 (as well as #291 )

Patch:
diff --git a/examples-parent/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java b/examples-parent/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
index 3869ea1..2ef00f3 100644
--- a/examples-parent/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
+++ b/examples-parent/send-money/src/main/java/org/interledger/examples/SendMoneyExample.java
@@ -12,6 +12,7 @@ import org.interledger.link.http.auth.SimpleBearerTokenSupplier;
 import org.interledger.spsp.PaymentPointer;
 import org.interledger.spsp.StreamConnectionDetails;
 import org.interledger.spsp.client.rust.InterledgerRustNodeClient;
+import org.interledger.stream.Denominations;
 import org.interledger.stream.SendMoneyResult;
 import org.interledger.core.SharedSecret;
 import org.interledger.stream.sender.SimpleStreamSender;
@@ -62,7 +63,8 @@ public class SendMoneyExample {
 
     // Send payment using STREAM
     SendMoneyResult result = simpleStreamSender.sendMoney(SharedSecret.of(connectionDetails.sharedSecret().value()),
-        SENDER_ADDRESS, connectionDetails.destinationAddress(), UnsignedLong.valueOf(100000), Duration.ofMillis(30000))
+        SENDER_ADDRESS, connectionDetails.destinationAddress(), UnsignedLong.valueOf(100000), Denominations.XRP,
+        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 a41a1f2..65e553d 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
@@ -18,14 +18,19 @@ import org.interledger.core.InterledgerResponsePacket;
 import org.interledger.core.SharedSecret;
 import org.interledger.encoding.asn.framework.CodecContext;
 import org.interledger.link.Link;
+import org.interledger.stream.Denomination;
+import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
 import org.interledger.stream.StreamConnection;
 import org.interledger.stream.StreamConnectionClosedException;
 import org.interledger.stream.StreamConnectionId;
 import org.interledger.stream.StreamPacket;
 import org.interledger.stream.StreamUtils;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
 import org.interledger.stream.crypto.JavaxStreamEncryptionService;
 import org.interledger.stream.crypto.StreamEncryptionService;
+import org.interledger.stream.frames.ConnectionAssetDetailsFrame;
 import org.interledger.stream.frames.ConnectionCloseFrame;
 import org.interledger.stream.frames.ConnectionNewAddressFrame;
 import org.interledger.stream.frames.StreamCloseFrame;
@@ -60,7 +65,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-
+import java.util.function.Function;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -162,28 +167,18 @@ public class SimpleStreamSender implements StreamSender {
       final SharedSecret sharedSecret,
       final InterledgerAddress sourceAddress,
       final InterledgerAddress destinationAddress,
-      final UnsignedLong amount
+      final UnsignedLong amount,
+      final Denomination denomination
   ) {
-    Objects.requireNonNull(destinationAddress);
-    Objects.requireNonNull(amount);
-
-    final StreamConnection streamConnection = this.streamConnectionManager.openConnection(
-        StreamConnectionId.from(destinationAddress, sharedSecret)
-    );
-
-    return new SendMoneyAggregator(
-        this.executorService,
-        streamConnection,
-        StreamCodecContextFactory.oer(),
-        this.link,
-        new AimdCongestionController(),
-        this.streamEncryptionService,
-        sharedSecret,
-        sourceAddress,
-        destinationAddress,
-        amount,
-        Optional.empty()
-    ).send();
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .sharedSecret(sharedSecret)
+        .destinationAddress(destinationAddress)
+        .sourceAddress(sourceAddress)
+        .amount(amount)
+        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .denomination(denomination)
+        .build();
+    return sendMoney(request);
   }
 
   @Override
@@ -192,13 +187,27 @@ public class SimpleStreamSender implements StreamSender {
       final InterledgerAddress sourceAddress,
       final InterledgerAddress destinationAddress,
       final UnsignedLong amount,
+      final Denomination denomination,
       final Duration timeout
   ) {
-    Objects.requireNonNull(destinationAddress);
-    Objects.requireNonNull(amount);
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .sharedSecret(sharedSecret)
+        .destinationAddress(destinationAddress)
+        .sourceAddress(sourceAddress)
+        .amount(amount)
+        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .denomination(denomination)
+        .timeout(timeout)
+        .build();
+    return sendMoney(request);
+  }
+
+  @Override
+  public CompletableFuture<SendMoneyResult> sendMoney(SendMoneyRequest request) {
+    Objects.requireNonNull(request);
 
     final StreamConnection streamConnection = this.streamConnectionManager.openConnection(
-        StreamConnectionId.from(destinationAddress, sharedSecret)
+        StreamConnectionId.from(request.destinationAddress(), request.sharedSecret())
     );
 
     return new SendMoneyAggregator(
@@ -208,11 +217,7 @@ public class SimpleStreamSender implements StreamSender {
         this.link,
         new AimdCongestionController(),
         this.streamEncryptionService,
-        sharedSecret,
-        sourceAddress,
-        destinationAddress,
-        amount,
-        Optional.of(timeout)
+        request
     ).send();
   }
 
@@ -275,6 +280,8 @@ public class SimpleStreamSender implements StreamSender {
     private final Link link;
     private final SharedSecret sharedSecret;
     private final Optional<Duration> timeout;
+    private final ExchangeRateCalculator exchangeRateCalculator;
+    private final Denomination senderDenomination;
 
     private final InterledgerAddress sourceAddress;
     private final InterledgerAddress destinationAddress;
@@ -291,22 +298,20 @@ public class SimpleStreamSender implements StreamSender {
     private final AtomicInteger numFulfilledPackets;
     private final AtomicInteger numRejectedPackets;
 
+    private Optional<Denomination> receiverDenomination = Optional.empty();
+
     /**
      * Required-args Constructor.
      *
-     * @param executorService         An {@link ExecutorService} for sending multiple STREAM frames in parallel.
-     * @param streamConnection        A {@link StreamConnection} that can be used to send packets with.
-     * @param streamCodecContext      A {@link CodecContext} that can encode and decode ASN.1 OER Stream packets and
-     *                                frames.
-     * @param link                    The {@link Link} used to send ILPv4 packets containing Stream packets.
-     * @param congestionController    A {@link CongestionController} that supports back-pressure for money streams.
-     * @param streamEncryptionService A {@link StreamEncryptionService} that allows for Stream packet encryption and
-     *                                decryption.
-     * @param sharedSecret            The shared secret negotiated between the sender and receiver during payment setup
-     *                                (e.g., using SPSP).
-     * @param sourceAddress           The {@link InterledgerAddress} of the payment sender.
-     * @param destinationAddress      The {@link InterledgerAddress} of the payment receiver.
-     * @param originalAmountToSend    The amount of units (in the senders units) to send to the receiver.
+     * @param executorService              An {@link ExecutorService} for sending multiple STREAM frames in parallel.
+     * @param streamConnection             A {@link StreamConnection} that can be used to send packets with.
+     * @param streamCodecContext           A {@link CodecContext} that can encode and decode ASN.1 OER Stream packets
+     *                                     and frames.
+     * @param link                         The {@link Link} used to send ILPv4 packets containing Stream packets.
+     * @param congestionController         A {@link CongestionController} that supports back-pressure for money streams.
+     * @param streamEncryptionService      A {@link StreamEncryptionService} that allows for Stream packet encryption
+     *                                     and decryption.
+     * @param request                      all relevant details about the money to send
      */
     SendMoneyAggregator(
         final ExecutorService executorService,
@@ -315,11 +320,7 @@ public class SimpleStreamSender implements StreamSender {
         final Link link,
         final CongestionController congestionController,
         final StreamEncryptionService streamEncryptionService,
-        final SharedSecret sharedSecret,
-        final InterledgerAddress sourceAddress,
-        final InterledgerAddress destinationAddress,
-        final UnsignedLong originalAmountToSend,
-        final Optional<Duration> timeout
+        final SendMoneyRequest request
     ) {
       this.executorService = Objects.requireNonNull(executorService);
       this.streamConnection = Objects.requireNonNull(streamConnection);
@@ -330,19 +331,22 @@ public class SimpleStreamSender implements StreamSender {
       this.congestionController = Objects.requireNonNull(congestionController);
       this.shouldSendSourceAddress = new AtomicBoolean(true);
 
-      this.sharedSecret = Objects.requireNonNull(sharedSecret);
-      this.sourceAddress = Objects.requireNonNull(sourceAddress);
-      this.destinationAddress = Objects.requireNonNull(destinationAddress);
+      this.sharedSecret = request.sharedSecret();
+      this.sourceAddress = request.sourceAddress();
+      this.destinationAddress = request.destinationAddress();
 
-      this.originalAmountToSend = new AtomicReference<>(originalAmountToSend);
-      this.amountLeftToSend = new AtomicReference<>(originalAmountToSend);
+      this.originalAmountToSend = new AtomicReference<>(request.amount());
+      this.amountLeftToSend = new AtomicReference<>(request.amount());
       this.sentAmount = new AtomicReference<>(UnsignedLong.ZERO);
       this.deliveredAmount = new AtomicReference<>(UnsignedLong.ZERO);
 
       this.numFulfilledPackets = new AtomicInteger(0);
       this.numRejectedPackets = new AtomicInteger(0);
 
-      this.timeout = Objects.requireNonNull(timeout);
+      this.timeout = request.timeout();
+
+      this.exchangeRateCalculator = request.exchangeRateCalculator();
+      this.senderDenomination = request.denomination();
     }
 
     /**
@@ -355,6 +359,26 @@ public class SimpleStreamSender implements StreamSender {
       Objects.requireNonNull(destinationAddress);
       Objects.requireNonNull(originalAmountToSend);
 
+      Instant startPreflight = Instant.now();
+      try {
+        receiverDenomination = preflightCheck();
+      } catch (StreamConnectionClosedException e) {
+        return CompletableFuture.completedFuture(SendMoneyResult.builder()
+            .sendMoneyDuration(Duration.between(startPreflight, Instant.now()))
+            .numRejectPackets(1)
+            .numFulfilledPackets(0)
+            .amountDelivered(UnsignedLong.ZERO)
+            .amountSent(UnsignedLong.ZERO)
+            .originalAmount(originalAmountToSend.get())
+            .amountLeftToSend(originalAmountToSend.get())
+            .build());
+      } catch (Exception e) {
+        logger.warn("Preflight check failed", e);
+      }
+      // A separate executor is needed for overall call to sendMoneyPacketized otherwise a livelock can occur.
+      // Using a shared executor could cause sendMoneyPacketized to internally get blocked from submitting tasks
+      // because the shared executor is already blocked waiting on the results of the call here to sendMoneyPacketized
+      ExecutorService sendMoneyExecutor = Executors.newSingleThreadExecutor();
       final Instant start = Instant.now();
       // All futures will run here using the Cached Executor service.
       return CompletableFuture
@@ -371,8 +395,9 @@ public class SimpleStreamSender implements StreamSender {
                 .numRejectPackets(numRejectedPackets.get())
                 .sendMoneyDuration(Duration.between(start, Instant.now()))
                 .build();
-          }, executorService)
+          }, sendMoneyExecutor)
           .whenComplete(($, error) -> {
+            sendMoneyExecutor.shutdown();
             if (error != null) {
               logger.error("SendMoney Stream failed: " + error.getMessage(), error);
             }
@@ -386,6 +411,71 @@ public class SimpleStreamSender implements StreamSender {
      * TODO: See https://github.com/hyperledger/quilt/issues/308 to determine when the Stream and/or Connection should
      * be closed.
      */
+    @VisibleForTesting
+    Optional<Denomination> preflightCheck() throws StreamConnectionClosedException {
+      // Load up the STREAM packet
+      final UnsignedLong sequence;
+      try {
+        sequence = this.streamConnection.nextSequence();
+      } catch (StreamConnectionClosedException e) {
+        // The Connection is closed, so we can't send anything more on it.
+        logger.warn(
+            "Unable to send more packets on a closed StreamConnection. streamConnection={} error={}",
+            streamConnection, e
+        );
+        throw e;
+      }
+
+      final List<StreamFrame> frames = Lists.newArrayList(
+          StreamMoneyFrame.builder()
+              // This aggregator supports only a simple stream-id, which is one.
+              .streamId(UnsignedLong.ONE)
+              .shares(UnsignedLong.ONE)
+              .build()
+      );
+
+      frames.add(ConnectionNewAddressFrame.builder()
+          .sourceAddress(sourceAddress)
+          .build()
+      );
+
+      final StreamPacket streamPacket = StreamPacket.builder()
+          .interledgerPacketType(InterledgerPacketType.PREPARE)
+          .prepareAmount(UnsignedLong.ZERO)
+          .sequence(sequence)
+          .frames(frames)
+          .build();
+
+      // Create the ILP Prepare packet
+      final byte[] streamPacketData = this.toEncrypted(sharedSecret, streamPacket);
+      final InterledgerCondition executionCondition;
+      executionCondition = generatedFulfillableFulfillment(sharedSecret, streamPacketData).getCondition();
+
+      final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
+          .destination(destinationAddress)
+          .amount(UnsignedLong.ZERO)
+          .executionCondition(executionCondition)
+          .expiresAt(Instant.now().plusSeconds(30L))
+          .data(streamPacketData)
+          .build();
+
+      InterledgerResponsePacket responsePacket = link.sendPacket(preparePacket);
+
+      final Function<InterledgerResponsePacket, Optional<Denomination>> readDetails = (p) -> {
+        final StreamPacket packet = this.fromEncrypted(sharedSecret, p.getData());
+        return packet.frames().stream()
+            .filter(f -> f.streamFrameType() == StreamFrameType.ConnectionAssetDetails)
+            .findFirst()
+            .map(f -> (ConnectionAssetDetailsFrame) f)
+            .map(f -> Denomination.builder().from(f.sourceDenomination()).build());
+      };
+
+      return responsePacket.map(
+          fulfillPacket -> readDetails.apply(fulfillPacket),
+          rejectPacket -> readDetails.apply(rejectPacket)
+      );
+    }
+
     private void sendMoneyPacketized() {
 
       final AtomicBoolean timeoutReached = new AtomicBoolean(false);
@@ -435,21 +525,13 @@ public class SimpleStreamSender implements StreamSender {
                 .build()
         );
 
-        // This isn't perfectly synchronized (it will be true until the first fulfill comes back), but it's OK if
-        // potentially many ILPv4 packets are sent out with this frame because it will always be the same "new" source
-        // address, so the receiver will merely update to the same address a few times, which is fine.
-        if (this.shouldSendSourceAddress.get()) {
-          frames.add(ConnectionNewAddressFrame.builder()
-              .sourceAddress(sourceAddress)
-              .build()
-          );
-        }
+        UnsignedLong receiverMinimum =
+            exchangeRateCalculator.calculate(amountToSend, senderDenomination, receiverDenomination);
 
         final StreamPacket streamPacket = StreamPacket.builder()
             .interledgerPacketType(InterledgerPacketType.PREPARE)
             // If the STREAM packet is sent on an ILP Prepare, this represents the minimum the receiver should accept.
-            // TODO: enforce min exchange rate?
-            .prepareAmount(UnsignedLong.ZERO)
+            .prepareAmount(receiverMinimum)
             .sequence(sequence)
             .frames(frames)
             .build();
@@ -477,41 +559,10 @@ public class SimpleStreamSender implements StreamSender {
             // controller to not reflect what we've actually scheduled to run, resulting in the loop
             // breaking prematurely
             congestionController.prepare(amountToSend);
-            try {
-              executorService.submit(() -> {
-                if (!timeoutReached.get()) {
-                  try {
-                    InterledgerResponsePacket responsePacket = link.sendPacket(preparePacket);
-                    responsePacket.handle(
-                        fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket),
-                        rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket)
-                    );
-                  } catch (Exception e) {
-                    logger.error("Link send failed. preparePacket={}", preparePacket, e);
-                    congestionController.reject(amountToSend, InterledgerRejectPacket.builder()
-                        .code(InterledgerErrorCode.F00_BAD_REQUEST)
-                        .message(
-                            String.format("Link send failed. preparePacket=%s error=%s", preparePacket, e.getMessage())
-                        )
-                        .build());
-                    this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.plus(amountToSend));
-                  }
-                }
-              });
-            } catch (RejectedExecutionException e) {
-              // If we get here, it means the task was unable to be scheduled, so we need to unwind the congestion
-              // controller to prevent deadlock.
-              congestionController.reject(amountToSend, InterledgerRejectPacket.builder()
-                  .code(InterledgerErrorCode.F00_BAD_REQUEST)
-                  .message(
-                      String.format("Unable to schedule sendMoney task. preparePacket=%s error=%s", preparePacket,
-                          e.getMessage())
-                  )
-                  .build());
-              throw e;
-            }
+            schedule(timeoutReached, preparePacket, streamPacket, amountToSend);
           } else {
             logger.error("SoldierOn runLoop had more tasks to schedule but was timed-out");
+            continue;
           }
         } catch (Exception e) {
           // Retry this amount on the next run...
@@ -522,6 +573,45 @@ public class SimpleStreamSender implements StreamSender {
       timeoutMonitor.shutdownNow();
     }
 
+    @VisibleForTesting
+    void schedule(AtomicBoolean timeoutReached, InterledgerPreparePacket preparePacket, StreamPacket streamPacket,
+                  UnsignedLong amountToSend) {
+      try {
+        executorService.submit(() -> {
+          if (!timeoutReached.get()) {
+            try {
+              InterledgerResponsePacket responsePacket = link.sendPacket(preparePacket);
+              responsePacket.handle(
+                  fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket),
+                  rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, numRejectedPackets,
+                      amountLeftToSend, congestionController)
+              );
+            } catch (Exception e) {
+              logger.error("Link send failed. preparePacket={}", preparePacket, e);
+              congestionController.reject(amountToSend, InterledgerRejectPacket.builder()
+                  .code(InterledgerErrorCode.F00_BAD_REQUEST)
+                  .message(
+                      String.format("Link send failed. preparePacket=%s error=%s", preparePacket, e.getMessage())
+                  )
+                  .build());
+              this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.plus(amountToSend));
+            }
+          }
+        });
+      } catch (RejectedExecutionException e) {
+        // If we get here, it means the task was unable to be scheduled, so we need to unwind the congestion
+        // controller to prevent deadlock.
+        congestionController.reject(amountToSend, InterledgerRejectPacket.builder()
+            .code(InterledgerErrorCode.F00_BAD_REQUEST)
+            .message(
+                String.format("Unable to schedule sendMoney task. preparePacket=%s error=%s", preparePacket,
+                    e.getMessage())
+            )
+            .build());
+        throw e;
+      }
+    }
+
     @VisibleForTesting
     boolean soldierOn(boolean timeoutReached) {
       // if money in flight, always soldier on
@@ -631,23 +721,29 @@ public class SimpleStreamSender implements StreamSender {
     void handleReject(
         final InterledgerPreparePacket originalPreparePacket,
         final StreamPacket originalStreamPacket,
-        final InterledgerRejectPacket rejectPacket
+        final InterledgerRejectPacket rejectPacket,
+        final AtomicInteger numRejectedPackets,
+        final AtomicReference<UnsignedLong> amountLeftToSend,
+        final CongestionController congestionController
     ) {
       Objects.requireNonNull(originalPreparePacket);
       Objects.requireNonNull(originalStreamPacket);
       Objects.requireNonNull(rejectPacket);
+      Objects.requireNonNull(numRejectedPackets);
+      Objects.requireNonNull(amountLeftToSend);
+      Objects.requireNonNull(congestionController);
 
       final UnsignedLong amountToSend = originalPreparePacket.getAmount();
-      this.numRejectedPackets.getAndIncrement();
-      this.amountLeftToSend.getAndUpdate(currentAmount -> currentAmount.plus(amountToSend));
-      this.congestionController.reject(amountToSend, rejectPacket);
+      numRejectedPackets.getAndIncrement();
+      amountLeftToSend.getAndUpdate(currentAmount -> currentAmount.plus(amountToSend));
+      congestionController.reject(amountToSend, rejectPacket);
 
       logger.debug(
           "Prepare with amount {} was rejected with code: {} ({} left to send). originalPreparePacket={} "
               + "originalStreamPacket={} rejectPacket={}",
           amountToSend,
           rejectPacket.getCode().getCode(),
-          this.amountLeftToSend.get(),
+          amountLeftToSend.get(),
           originalPreparePacket,
           originalStreamPacket,
           rejectPacket
@@ -740,7 +836,8 @@ public class SimpleStreamSender implements StreamSender {
 
       link.sendPacket(preparePacket).handle(
           fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket),
-          rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket)
+          rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, numRejectedPackets, amountLeftToSend,
+              congestionController)
       );
 
       // Mark the streamConnection object as closed if the caller supplied a ConnectionCloseFrame
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/StreamSender.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/StreamSender.java
index fa31302..828a0d3 100644
--- a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/StreamSender.java
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/StreamSender.java
@@ -1,11 +1,12 @@
 package org.interledger.stream.sender;
 
+import com.google.common.primitives.UnsignedLong;
 import org.interledger.core.InterledgerAddress;
 import org.interledger.core.SharedSecret;
+import org.interledger.stream.Denomination;
+import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
 
-import com.google.common.primitives.UnsignedLong;
-
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 
@@ -28,6 +29,7 @@ public interface StreamSender {
    * @param sourceAddress      The {@link InterledgerAddress} of the source of this payment.
    * @param destinationAddress The {@link InterledgerAddress} of the receiver of this payment.
    * @param amount             An {@link UnsignedLong} containing the amount of this payment.
+   * @param denomination       The denomination of the amount to be sent
    *
    * @return A {@link CompletableFuture} that, once complete, will contain either an error or an instance of {@link
    *     SendMoneyResult} that displays the result of this "send money" request.
@@ -36,12 +38,13 @@ public interface StreamSender {
       final SharedSecret sharedSecret,
       final InterledgerAddress sourceAddress,
       final InterledgerAddress destinationAddress,
-      final UnsignedLong amount
-  );
+      final UnsignedLong amount,
+      final Denomination denomination
+      );
 
   /**
    * <p>Send "money" (i.e., some unit of value) from a source ILP address to a destination address, preventing any
-   * unsent STREAM packets from becoming encqueed if the payment timeout has been exceeded.</p>
+   * unsent STREAM packets from becoming enqueued if the payment timeout has been exceeded.</p>
    *
    * <p>Note that, per https://github.com/hyperledger/quilt/issues/242, as of the publication of this client,
    * connectors will reject ILP packets that exceed 32kb (though there is no hard rule that more than 32kb will not be
@@ -52,6 +55,7 @@ public interface StreamSender {
    * @param sourceAddress      The {@link InterledgerAddress} of the source of this payment.
    * @param destinationAddress The {@link InterledgerAddress} of the receiver of this payment.
    * @param amount             An {@link UnsignedLong} containing the amount of this payment.
+   * @param denomination       The denomination of the amount to be sent
    * @param timeout            A {@link Duration} to wait before no longer scheduling any off more requests to send
    *                           stream packets for this payment. This is an important distinction as compared to a
    *                           traditional `hard timeout` where processing might end immediately upon timeout. Instead,
@@ -66,7 +70,21 @@ public interface StreamSender {
       final InterledgerAddress sourceAddress,
       final InterledgerAddress destinationAddress,
       final UnsignedLong amount,
+      final Denomination denomination,
       final Duration timeout
   );
 
+  /**
+   * <p>Send "money" (i.e., some unit of value) from a source ILP address to a destination address, preventing any
+   * unsent STREAM packets from becoming enqueued if the optional payment timeout has been exceeded.</p>
+   *
+   * <p>Note that, per https://github.com/hyperledger/quilt/issues/242, as of the publication of this client,
+   * connectors will reject ILP packets that exceed 32kb (though there is no hard rule that more than 32kb will not be
+   * supported in the future.</p>
+   *
+   * @param request all relevant details about the money to send
+   * @return results of the request to send money
+   */
+  CompletableFuture<SendMoneyResult> sendMoney(SendMoneyRequest request);
+
 }
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/Denominations.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/Denominations.java
new file mode 100644
index 0000000..d850fe5
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/Denominations.java
@@ -0,0 +1,32 @@
+package org.interledger.stream;
+
+import org.interledger.stream.Denomination;
+
+public final class Denominations {
+  private Denominations(){}
+
+  public static final Denomination XRP = Denomination.builder()
+      .assetCode("XRP")
+      .assetScale((short) 6)
+      .build();
+
+  public static final Denomination USD = Denomination.builder()
+      .assetCode("USD")
+      .assetScale((short) 0)
+      .build();
+
+  public static final Denomination US_CENTS = Denomination.builder()
+      .assetCode("USD")
+      .assetScale((short) 2)
+      .build();
+
+  public static final Denomination EUR = Denomination.builder()
+      .assetCode("EUR")
+      .assetScale((short) 0)
+      .build();
+
+  public static final Denomination EUR_CENTS = Denomination.builder()
+      .assetCode("EUR")
+      .assetScale((short) 2)
+      .build();
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyRequest.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyRequest.java
new file mode 100644
index 0000000..6b411e2
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/SendMoneyRequest.java
@@ -0,0 +1,65 @@
+package org.interledger.stream;
+
+import com.google.common.primitives.UnsignedLong;
+import org.interledger.core.Immutable;
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.SharedSecret;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+
+import java.time.Duration;
+import java.util.Optional;
+
+@Immutable
+public interface SendMoneyRequest {
+
+  static SendMoneyRequestBuilder builder() { return new SendMoneyRequestBuilder(); }
+
+  /**
+   *
+   * @return A {@link SharedSecret} known only to the sender and receiver, negotiated using
+   *         some higher-level protocol (e.g., SPSP or something else).
+   */
+  SharedSecret sharedSecret();
+
+  /**
+   *
+   * @return The {@link InterledgerAddress} of the source of this payment.
+   */
+  InterledgerAddress sourceAddress();
+
+  /**
+   *
+   * @return The {@link InterledgerAddress} of the receiver of this payment.
+   */
+  InterledgerAddress destinationAddress();
+
+  /**
+   *
+   * @return An {@link UnsignedLong} containing the amount of this payment.
+   */
+  UnsignedLong amount();
+
+  /**
+   *
+   * @return
+   */
+  Denomination denomination();
+
+  /**
+   *
+   * @return A {@link Duration} to wait before no longer scheduling any off more requests
+   *         to send stream packets for this payment. This is an important distinction as
+   *         compared to a traditional `hard timeout` where processing might end
+   *         immediately upon timeout. Instead, this timeout should be considered to be a
+   *         `soft timeout`, because the sender will wait for any in-flight packetized
+   *         payments to be sent before exiting out of it run-loop.
+   */
+  Optional<Duration> timeout();
+
+  /**
+   *
+   * @return a calculator to figure out how much money should arrive on the other side of a given payment.
+   */
+  ExchangeRateCalculator exchangeRateCalculator();
+
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/ExchangeRateCalculator.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/ExchangeRateCalculator.java
new file mode 100644
index 0000000..009ca19
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/ExchangeRateCalculator.java
@@ -0,0 +1,36 @@
+package org.interledger.stream.calculators;
+
+import com.google.common.primitives.UnsignedLong;
+import org.interledger.stream.Denomination;
+
+import java.math.BigDecimal;
+import java.util.Optional;
+
+public interface ExchangeRateCalculator {
+
+  /**
+   * Calculate the minimum amount a receiver should accept based on the exchange rate applied to the sendAmount.
+   * @param sendAmount                   amount to send.
+   * @param sendDenomination             asset code and scale of the amount to send.
+   * @param expectedReceivedDenomination asset code and scale to apply to the amount the receiver should receive
+   *                                     when computing the total for an exchange rate. Optional if denomination of
+   *                                     receiver is unknown. ExchangeRateCalculator implementations should throw a
+   *                                     {@link NoExchangeRateException}
+   * @return                             the minimum amount the receiver should accept.
+   * @throws NoExchangeRateException if exchange rate could not be calculated
+   */
+  UnsignedLong calculate(UnsignedLong sendAmount, Denomination sendDenomination,
+                         Optional<Denomination> expectedReceivedDenomination)
+    throws NoExchangeRateException;
+
+  /**
+   * Convenience method to compute the BigDecimal representation of an amount for a given asset scale
+   * @param amount the non-decimal version of the amount
+   * @param denomination definition of the scale
+   * @return the scaled decimal amount
+   */
+  default BigDecimal scaled(UnsignedLong amount, Denomination denomination) {
+    BigDecimal scale = BigDecimal.ONE.scaleByPowerOfTen(denomination.assetScale());
+    return new BigDecimal(amount.bigIntegerValue()).divide(scale);
+  }
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoExchangeRateException.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoExchangeRateException.java
new file mode 100644
index 0000000..6c3e874
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoExchangeRateException.java
@@ -0,0 +1,13 @@
+package org.interledger.stream.calculators;
+
+/**
+ * Thrown when an exchange rate could not be determined. This could happen if the destination denomination is
+ * unknown or if an exchange rate cannot be found to the destination denomination.
+ */
+public class NoExchangeRateException extends RuntimeException {
+
+  public NoExchangeRateException(String message) {
+    super(message);
+  }
+
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoOpExchangeRateCalculator.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoOpExchangeRateCalculator.java
new file mode 100644
index 0000000..cdfa9ab
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/calculators/NoOpExchangeRateCalculator.java
@@ -0,0 +1,15 @@
+package org.interledger.stream.calculators;
+
+import com.google.common.primitives.UnsignedLong;
+import org.interledger.stream.Denomination;
+import java.util.Optional;
+
+public class NoOpExchangeRateCalculator implements ExchangeRateCalculator {
+
+  @Override
+  public UnsignedLong calculate(UnsignedLong sendAmount, Denomination sendDenomination,
+                                Optional<Denomination> expectedReceivedDenomination) {
+    return UnsignedLong.ZERO;
+  }
+
+}


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 62ff84b..7867df2 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
@@ -2,33 +2,59 @@ package org.interledger.stream.sender;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerCondition;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerFulfillPacket;
+import org.interledger.core.InterledgerFulfillment;
+import org.interledger.core.InterledgerPacketType;
+import org.interledger.core.InterledgerPreparePacket;
 import org.interledger.core.InterledgerRejectPacket;
 import org.interledger.core.SharedSecret;
 import org.interledger.encoding.asn.framework.CodecContext;
 import org.interledger.link.Link;
+import org.interledger.stream.Denomination;
+import org.interledger.stream.Denominations;
+import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.StreamConnection;
 import org.interledger.stream.StreamConnectionClosedException;
+import org.interledger.stream.StreamConnectionId;
+import org.interledger.stream.StreamPacket;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
 import org.interledger.stream.crypto.StreamEncryptionService;
+import org.interledger.stream.frames.StreamMoneyFrame;
 import org.interledger.stream.sender.SimpleStreamSender.SendMoneyAggregator;
 
 import com.google.common.primitives.UnsignedLong;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.junit.rules.Timeout;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
+import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Unit tests for {@link SendMoneyAggregator}.
@@ -39,6 +65,9 @@ public class SendMoneyAggregatorTest {
   @Rule
   public Timeout globalTimeout = Timeout.seconds(5);
 
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
   @Mock
   private CodecContext streamCodecContextMock;
   @Mock
@@ -65,13 +94,19 @@ public class SendMoneyAggregatorTest {
     when(streamEncryptionServiceMock.encrypt(any(), any())).thenReturn(new byte[32]);
     when(streamEncryptionServiceMock.decrypt(any(), any())).thenReturn(new byte[32]);
     when(linkMock.sendPacket(any())).thenReturn(mock(InterledgerRejectPacket.class));
-
-    final ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
-
+    ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .sharedSecret(sharedSecret)
+        .sourceAddress(sourceAddress)
+        .destinationAddress(destinationAddress)
+        .amount(originalAmountToSend)
+        .timeout(Optional.of(Duration.ofSeconds(60)))
+        .denomination(Denominations.XRP)
+        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .build();
     this.sendMoneyAggregator = new SendMoneyAggregator(
         executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-        streamEncryptionServiceMock, sharedSecret, sourceAddress, destinationAddress, originalAmountToSend,
-        Optional.of(Duration.ofSeconds(60))
+        streamEncryptionServiceMock, request
     );
   }
 
@@ -84,8 +119,8 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 0 Link calls since close Connection is not called automatically
-    Mockito.verifyNoMoreInteractions(linkMock);
+    // Expect 1 Link call due to preflight check
+    Mockito.verify(linkMock, times(1)).sendPacket(any());
   }
 
   @Test
@@ -97,8 +132,8 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 0 Link calls since close Connection is not called automatically
-    Mockito.verifyNoMoreInteractions(linkMock);
+    // Expect 1 Link call due to preflight check
+    Mockito.verify(linkMock, times(1)).sendPacket(any());
   }
 
   @Test
@@ -110,8 +145,8 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 0 Link calls since close Connection is not called automatically
-    Mockito.verifyNoMoreInteractions(linkMock);
+    // Expect 1 Link call due to preflight check
+    Mockito.verify(linkMock, times(1)).sendPacket(any());
   }
 
   @Test
@@ -124,10 +159,111 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 0 Link calls since close Connection is not called automatically
+    // Expect 1 Link call due to preflight check
+    Mockito.verify(linkMock, times(1)).sendPacket(any());
+  }
+
+  @Test
+  public void sendMoneyFailsPreflightSequenceIncrement()
+      throws ExecutionException, InterruptedException, StreamConnectionClosedException {
+
+    setSoldierOnBooleans(false, true, true, false);
+    when(streamConnectionMock.nextSequence())
+        .thenThrow(new StreamConnectionClosedException(StreamConnectionId.of("whoops")));
+
+    sendMoneyAggregator.send().get();
+
+    // Expect 0 link calls
     Mockito.verifyNoMoreInteractions(linkMock);
   }
 
+  @Test
+  public void sendMoneyWhenSequenceCannotIncrement()
+      throws ExecutionException, InterruptedException, StreamConnectionClosedException {
+
+    setSoldierOnBooleans(false, true, true, false);
+    when(streamConnectionMock.nextSequence()).thenAnswer(new Answer<UnsignedLong>() {
+      AtomicInteger counter = new AtomicInteger(0);
+
+      @Override
+      public UnsignedLong answer(InvocationOnMock invocationOnMock) throws Throwable {
+        int i = counter.incrementAndGet();
+        if (i > 1) {
+          throw new StreamConnectionClosedException(StreamConnectionId.of("whoops"));
+        }
+        return UnsignedLong.ONE;
+      }
+    });
+
+    sendMoneyAggregator.send().get();
+
+    // Expect 1 Link call due to preflight check
+    Mockito.verify(linkMock, times(1)).sendPacket(any());
+  }
+
+  @Test
+  public void failureToSchedulePutsMoneyBack() {
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .sharedSecret(sharedSecret)
+        .sourceAddress(sourceAddress)
+        .destinationAddress(destinationAddress)
+        .amount(originalAmountToSend)
+        .timeout(Optional.of(Duration.ofSeconds(60)))
+        .denomination(Denominations.XRP)
+        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .build();
+    ExecutorService executor = mock(ExecutorService.class);
+    this.sendMoneyAggregator = new SendMoneyAggregator(
+        executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
+        streamEncryptionServiceMock, request);
+
+    when(executor.submit(any(Runnable.class))).thenThrow(new RejectedExecutionException());
+
+    InterledgerPreparePacket prepare = samplePreparePacket();
+    InterledgerRejectPacket expectedReject = InterledgerRejectPacket.builder()
+        .code(InterledgerErrorCode.F00_BAD_REQUEST)
+        .message(
+            String.format("Unable to schedule sendMoney task. preparePacket=%s error=%s", prepare,
+                "java.util.concurrent.RejectedExecutionException")
+        )
+        .build();
+
+    expectedException.expect(RejectedExecutionException.class);
+    sendMoneyAggregator.schedule(new AtomicBoolean(false), prepare, sampleStreamPacket(), UnsignedLong.ONE);
+    verify(congestionControllerMock, times(1)).reject(UnsignedLong.ONE, expectedReject);
+  }
+
+  @Test
+  public void preflightCheckFindsNoDenomination() throws Exception {
+    when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
+    when(linkMock.sendPacket(any())).thenReturn(sampleFulfillPacket());
+    StreamPacket streamPacket = StreamPacket.builder().from(sampleStreamPacket())
+        .addFrames(StreamMoneyFrame.builder()
+            .shares(UnsignedLong.ONE)
+            .streamId(UnsignedLong.ONE)
+            .build())
+        .build();
+    when(streamCodecContextMock.read(any(), any())).thenReturn(streamPacket);
+
+    Optional<Denomination> denomination = sendMoneyAggregator.preflightCheck();
+    assertThat(denomination).isEmpty();
+  }
+
+  @Test
+  public void preflightCheckRejects() throws Exception{
+    when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
+    when(linkMock.sendPacket(any())).thenReturn(sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR));
+    StreamPacket streamPacket = StreamPacket.builder().from(sampleStreamPacket())
+        .addFrames(StreamMoneyFrame.builder()
+            .shares(UnsignedLong.ONE)
+            .streamId(UnsignedLong.ONE)
+            .build())
+        .build();
+    when(streamCodecContextMock.read(any(), any())).thenReturn(streamPacket);
+    Optional<Denomination> denomination = sendMoneyAggregator.preflightCheck();
+    assertThat(denomination).isEmpty();
+  }
+
   @Test
   public void soldierOn() {
     // if money in flight, always soldier on
@@ -185,6 +321,81 @@ public class SendMoneyAggregatorTest {
     assertThat(sendMoneyAggregator.soldierOn(true)).isTrue();
   }
 
+  @Test
+  public void toEncrypted() throws Exception {
+    doThrow(new IOException()).when(streamCodecContextMock).write(any(), any());
+    expectedException.expect(StreamSenderException.class);
+    StreamPacket packet = sampleStreamPacket();
+    sendMoneyAggregator.toEncrypted(sharedSecret, packet);
+  }
+
+  @Test
+  public void fromEncrypted() throws Exception {
+    doThrow(new IOException()).when(streamCodecContextMock).read(any(), any());
+    expectedException.expect(StreamSenderException.class);
+    sendMoneyAggregator.fromEncrypted(sharedSecret, new byte[0]);
+  }
+
+  @Test
+  public void handleRejectHatesNullPrepare() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
+        new AtomicReference<>(), congestionControllerMock);
+  }
+
+  @Test
+  public void handleRejectHatesNullStreamPacket() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(samplePreparePacket(), null,
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
+        new AtomicReference<>(), congestionControllerMock);
+  }
+
+  @Test
+  public void handleRejectHatesNullReject() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(), null, new AtomicInteger(),
+        new AtomicReference<UnsignedLong>(), congestionControllerMock);
+  }
+
+  @Test
+  public void handleRejectHatesNullNumReject() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(), sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR),
+        null, new AtomicReference<UnsignedLong>(), congestionControllerMock);
+  }
+
+  @Test
+  public void handleRejectHatesNullAmountLeftToSend() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
+        null, congestionControllerMock);
+  }
+
+  @Test
+  public void handleRejectHatesNullCongestionController() {
+    expectedException.expect(NullPointerException.class);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
+        new AtomicReference<UnsignedLong>(), null);
+  }
+
+  @Test
+  public void handleReject() {
+    AtomicInteger numReject = new AtomicInteger(0);
+    AtomicReference<UnsignedLong> amountLeftToSend = new AtomicReference<>(UnsignedLong.ONE);
+    InterledgerPreparePacket prepare = samplePreparePacket();
+    InterledgerRejectPacket reject = sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR);
+    sendMoneyAggregator.handleReject(prepare, sampleStreamPacket(),
+        reject, numReject, amountLeftToSend,
+        congestionControllerMock);
+    assertThat(numReject.get()).isEqualTo(1);
+    assertThat(amountLeftToSend.get().intValue()).isEqualTo(2);
+    verify(congestionControllerMock, times(1)).reject(UnsignedLong.ONE, reject);
+  }
+
   /**
    * Helper method to set the soldierOn mock values for clearer test coverage.
    */
@@ -200,4 +411,38 @@ public class SendMoneyAggregatorTest {
       sendMoneyAggregator.setAmountSentForTesting(UnsignedLong.valueOf(10L));
     }
   }
+
+  private StreamPacket sampleStreamPacket() {
+    return StreamPacket.builder()
+        .prepareAmount(UnsignedLong.ZERO)
+        .sequence(UnsignedLong.ZERO)
+        .interledgerPacketType(InterledgerPacketType.PREPARE)
+        .build();
+  }
+
+  private InterledgerPreparePacket samplePreparePacket() {
+    return InterledgerPreparePacket.builder()
+        .destination(destinationAddress)
+        .amount(UnsignedLong.ONE)
+        .expiresAt(Instant.now())
+        .executionCondition(InterledgerCondition.of(new byte[32]))
+        .build();
+  }
+
+  private InterledgerRejectPacket sampleRejectPacket(InterledgerErrorCode errorCode) {
+    return InterledgerRejectPacket.builder()
+        .code(errorCode)
+        .message("too many cooks!")
+        .triggeredBy(destinationAddress)
+        .data(new byte[0])
+        .build();
+  }
+
+  private InterledgerFulfillPacket sampleFulfillPacket() {
+    return InterledgerFulfillPacket.builder()
+        .data(new byte[0])
+        .fulfillment(InterledgerFulfillment.of(new byte[32]))
+        .build();
+  }
+
 }
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
index 867d95f..8e6c7b5 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
@@ -2,6 +2,7 @@ package org.interledger.stream.sender;
 
 import static okhttp3.CookieJar.NO_COOKIES;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Fail.fail;
 
 import org.interledger.codecs.ilp.InterledgerCodecContextFactory;
 import org.interledger.core.InterledgerAddress;
@@ -21,7 +22,12 @@ import org.interledger.spsp.StreamConnectionDetails;
 import org.interledger.spsp.client.rust.Account;
 import org.interledger.spsp.client.rust.ImmutableAccount;
 import org.interledger.spsp.client.rust.InterledgerRustNodeClient;
+import org.interledger.stream.Denomination;
+import org.interledger.stream.Denominations;
+import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+import org.interledger.stream.calculators.NoExchangeRateException;
 import org.interledger.stream.crypto.JavaxStreamEncryptionService;
 
 import com.fasterxml.jackson.core.JsonGenerator;
@@ -56,8 +62,11 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -156,7 +165,7 @@ public class SimpleStreamSenderIT {
 
   /**
    * One call to {@link SimpleStreamSender#sendMoney(SharedSecret, InterledgerAddress, InterledgerAddress,
-   * UnsignedLong)} that involves a single packet for the entire payment.
+   * UnsignedLong, Denomination)} that involves a single packet for the entire payment.
    */
   @Test
   public void sendMoneySinglePacket() {
@@ -172,7 +181,8 @@ public class SimpleStreamSenderIT {
         SharedSecret.of(connectionDetails.sharedSecret().key()),
         SENDER_ADDRESS,
         connectionDetails.destinationAddress(),
-        paymentAmount
+        paymentAmount,
+        Denominations.XRP
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -185,8 +195,9 @@ public class SimpleStreamSenderIT {
 
   /**
    * In general, calling sendMoney using the same Connection (i.e., SharedSecret) in parallel should not be done.
-   * However, the implementation is smart enough to allow parallel requests to run in parallel, which this test
-   * validates.
+   * However, the implementation is smart enough to queue up parallel requests and only allow one to run at a time.
+   * However, sometimes waiting tasks will timeout, in which case a particular `sendMoney` may throw an exception. This
+   * test does not expect any exceptions.
    */
   @Test
   public void sendMoneyOnSameConnectionInParallel() {
@@ -206,7 +217,8 @@ public class SimpleStreamSenderIT {
           SharedSecret.of(connectionDetails.sharedSecret().key()),
           SENDER_ADDRESS,
           connectionDetails.destinationAddress(),
-          paymentAmount
+          paymentAmount,
+          Denominations.XRP
       );
       results.add(job);
     }
@@ -264,7 +276,8 @@ public class SimpleStreamSenderIT {
         SharedSecret.of(connectionDetails.sharedSecret().key()),
         SENDER_ADDRESS,
         connectionDetails.destinationAddress(),
-        paymentAmount
+        paymentAmount,
+        Denominations.XRP
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -349,6 +362,7 @@ public class SimpleStreamSenderIT {
             SENDER_ADDRESS,
             connectionDetails.destinationAddress(),
             paymentAmount,
+            Denominations.XRP,
             Duration.ofMillis(100)).join();
 
     assertThat(sendMoneyResult.successfulPayment()).isFalse();
@@ -356,6 +370,115 @@ public class SimpleStreamSenderIT {
     logger.info("Payment Sent: {}", sendMoneyResult);
   }
 
+  @Test(expected = NoExchangeRateException.class)
+  public void sendFailsIfNoExchangeRate() throws Throwable {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
+
+    StreamSender streamSender = new SimpleStreamSender(
+        new JavaxStreamEncryptionService(), link
+    );
+
+    String username = "sendFailsIfNoExchangeRate";
+    InterledgerAddress address = HOST_ADDRESS.with(username);
+    Account account = accountBuilder()
+        .username(username)
+        .ilpAddress(address)
+        .maxPacketAmount(BigInteger.valueOf(100))
+        .amountPerMinuteLimit(BigInteger.valueOf(1))
+        .packetsPerMinuteLimit(BigInteger.valueOf(1))
+        .build();
+
+    final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(account);
+
+    ExchangeRateCalculator noExchangeRateExceptionCalculator =
+        (sendAmount, sendDenomination, expectedReceivedDenomination) -> {
+      throw new NoExchangeRateException("Never stop never calculating");
+    };
+
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .amount(paymentAmount)
+        .denomination(Denominations.XRP)
+        .destinationAddress(connectionDetails.destinationAddress())
+        .exchangeRateCalculator(noExchangeRateExceptionCalculator)
+        .sharedSecret(connectionDetails.sharedSecret())
+        .sourceAddress(SENDER_ADDRESS)
+        .timeout(Duration.ofMillis(100))
+        .build();
+
+    try {
+      streamSender.sendMoney(request).join();
+    } catch (CompletionException e) {
+      throw e.getCause();
+    }
+  }
+
+  @Test
+  public void sendAcceptableDeliveredAmount() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(100);
+
+    StreamSender streamSender = new SimpleStreamSender(link);
+
+    String username = "deliveredAmountRejected";
+    InterledgerAddress address = HOST_ADDRESS.with(username);
+    Account account = accountBuilder()
+        .username(username)
+        .ilpAddress(address)
+        .build();
+
+    final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(account);
+
+    // this should always cause the packet to get rejected because the delivered amount will never exceed the sent amount
+    ExchangeRateCalculator reasonableCalculator =
+        (sendAmount, sendDenomination, expectedReceivedDenomination) -> sendAmount;
+
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .amount(paymentAmount)
+        .denomination(Denominations.XRP)
+        .destinationAddress(connectionDetails.destinationAddress())
+        .exchangeRateCalculator(reasonableCalculator)
+        .sharedSecret(connectionDetails.sharedSecret())
+        .sourceAddress(SENDER_ADDRESS)
+        .timeout(Duration.ofMillis(1000))
+        .build();
+
+    SendMoneyResult result = streamSender.sendMoney(request).join();
+    assertThat(result.amountDelivered()).isEqualTo(paymentAmount);
+    assertThat(result.successfulPayment()).isTrue();
+  }
+
+  @Test
+  public void sendFailsIfDeliveredAmountRejected() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(100);
+
+    StreamSender streamSender = new SimpleStreamSender(link);
+
+    String username = "deliveredAmountRejected";
+    InterledgerAddress address = HOST_ADDRESS.with(username);
+    Account account = accountBuilder()
+        .username(username)
+        .ilpAddress(address)
+        .build();
+
+    final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(account);
+
+    // this should always cause the packet to get rejected because the delivered amount will never exceed the sent amount
+    ExchangeRateCalculator unreasonableCalculator =
+        (sendAmount, sendDenomination, expectedReceivedDenomination) -> sendAmount.plus(UnsignedLong.valueOf(1));
+
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .amount(paymentAmount)
+        .denomination(Denominations.XRP)
+        .destinationAddress(connectionDetails.destinationAddress())
+        .exchangeRateCalculator(unreasonableCalculator)
+        .sharedSecret(connectionDetails.sharedSecret())
+        .sourceAddress(SENDER_ADDRESS)
+        .timeout(Duration.ofMillis(1000))
+        .build();
+
+    SendMoneyResult result = streamSender.sendMoney(request).join();
+    assertThat(result.amountDelivered()).isEqualTo(UnsignedLong.ZERO);
+  }
+
   @Test
   public void sendMoneyWithWrongLinkPassword() throws IOException {
     final String connectorAccountUsername = UUID.randomUUID().toString().replace("-", "");
@@ -403,6 +526,7 @@ public class SimpleStreamSenderIT {
         SENDER_ADDRESS,
         connectionDetails.destinationAddress(),
         paymentAmount,
+        Denominations.XRP,
         Duration.ofMillis(100L)
     ).whenComplete(($, error) -> {
       assertThat(error).isNotNull();
@@ -442,7 +566,8 @@ public class SimpleStreamSenderIT {
         SharedSecret.of(connectionDetails.sharedSecret().key()),
         SENDER_ADDRESS,
         connectionDetails.destinationAddress(),
-        paymentAmount
+        paymentAmount,
+        Denominations.XRP
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
index 224b870..c2859c6 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
@@ -1,7 +1,9 @@
 package org.interledger.stream.sender;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
+import com.google.common.primitives.UnsignedLong;
 import org.interledger.link.Link;
 import org.interledger.stream.crypto.StreamEncryptionService;
 
@@ -65,4 +67,14 @@ public class SimpleStreamSenderTests {
     expectedException.expect(NullPointerException.class);
     new SimpleStreamSender(streamEncryptionServiceMock, linkMock, null);
   }
+
+  @Test
+  public void connectionStats() {
+    SimpleStreamSender.ConnectionStatistics stats = SimpleStreamSender.ConnectionStatistics.builder()
+        .numFulfilledPackets(10)
+        .numRejectPackets(5)
+        .amountDelivered(UnsignedLong.ONE)
+        .build();
+    assertThat(stats.totalPackets()).isEqualTo(15);
+  }
 }
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/DenominationsTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/DenominationsTest.java
new file mode 100644
index 0000000..2a9b42d
--- /dev/null
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/DenominationsTest.java
@@ -0,0 +1,27 @@
+package org.interledger.stream;
+
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class DenominationsTest {
+
+  @Test
+  public void values() {
+    assertThat(Denominations.XRP)
+        .extracting("assetCode", "assetScale")
+        .containsExactly("XRP", (short) 6);
+    assertThat(Denominations.US_CENTS)
+        .extracting("assetCode", "assetScale")
+        .containsExactly("USD", (short) 2);
+    assertThat(Denominations.USD)
+        .extracting("assetCode", "assetScale")
+        .containsExactly("USD", (short) 0);
+    assertThat(Denominations.EUR_CENTS)
+        .extracting("assetCode", "assetScale")
+        .containsExactly("EUR", (short) 2);
+    assertThat(Denominations.EUR)
+        .extracting("assetCode", "assetScale")
+        .containsExactly("EUR", (short) 0);
+  }
+}
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/ExchangeRateCalculatorTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/ExchangeRateCalculatorTest.java
new file mode 100644
index 0000000..c92ed69
--- /dev/null
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/ExchangeRateCalculatorTest.java
@@ -0,0 +1,19 @@
+package org.interledger.stream.calculators;
+
+import com.google.common.primitives.UnsignedLong;
+import org.interledger.stream.Denominations;
+import org.junit.Test;
+
+import java.math.BigDecimal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+
+public class ExchangeRateCalculatorTest {
+
+  @Test
+  public void scaled() {
+    ExchangeRateCalculator calculator = spy(ExchangeRateCalculator.class);
+    assertThat(calculator.scaled(UnsignedLong.ONE, Denominations.US_CENTS)).isEqualTo(BigDecimal.valueOf(.01));
+  }
+}
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/NoOpExchangeRateCalculatorTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/NoOpExchangeRateCalculatorTest.java
new file mode 100644
index 0000000..269527e
--- /dev/null
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/calculators/NoOpExchangeRateCalculatorTest.java
@@ -0,0 +1,18 @@
+package org.interledger.stream.calculators;
+
+import com.google.common.primitives.UnsignedLong;
+import org.interledger.stream.Denominations;
+import org.junit.Test;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class NoOpExchangeRateCalculatorTest {
+
+  @Test
+  public void calculate() {
+    ExchangeRateCalculator calc = new NoOpExchangeRateCalculator();
+    assertThat(calc.calculate(UnsignedLong.ONE, Denominations.USD, Optional.of(Denominations.EUR))).isEqualTo(UnsignedLong.ZERO);
+  }
+}
diff --git a/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/SenderReceiverTest.java b/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/SenderReceiverTest.java
index 94b3de5..9a063aa 100644
--- a/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/SenderReceiverTest.java
+++ b/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/SenderReceiverTest.java
@@ -9,7 +9,10 @@ import org.interledger.core.SharedSecret;
 import org.interledger.link.Link;
 import org.interledger.spsp.StreamConnectionDetails;
 import org.interledger.stream.Denomination;
+import org.interledger.stream.Denominations;
+import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
 import org.interledger.stream.crypto.JavaxStreamEncryptionService;
 import org.interledger.stream.crypto.StreamEncryptionService;
 import org.interledger.stream.receiver.testutils.SimulatedIlpv4Network;
@@ -17,6 +20,7 @@ import org.interledger.stream.receiver.testutils.SimulatedPathConditions;
 import org.interledger.stream.sender.SimpleStreamSender;
 import org.interledger.stream.sender.StreamSender;
 
+import com.google.common.collect.Lists;
 import com.google.common.io.BaseEncoding;
 import com.google.common.primitives.UnsignedLong;
 import org.assertj.core.data.Offset;
@@ -26,7 +30,15 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
+import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 /**
  * A unit tests that simulates network connectivity between a sender and a receiver in order isolate and control various
@@ -61,12 +73,7 @@ public class SenderReceiverTest {
     final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
 
     final StreamConnectionDetails connectionDetails = leftStreamNode.getNewStreamConnectionDetails();
-    final SendMoneyResult sendMoneyResult = leftStreamNode.streamSender().sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        LEFT_SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount
-    ).join();
+    final SendMoneyResult sendMoneyResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
@@ -85,7 +92,8 @@ public class SenderReceiverTest {
         SharedSecret.of(connectionDetails.sharedSecret().key()),
         RIGHT_SENDER_ADDRESS,
         connectionDetails.destinationAddress(),
-        paymentAmount
+        paymentAmount,
+        Denominations.XRP
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -96,6 +104,20 @@ public class SenderReceiverTest {
     logger.info("Payment Sent: {}", sendMoneyResult);
   }
 
+  @Test
+  public void testSendCannotDetermineReceiverDenomination() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
+
+    final SendMoneyResult sendMoneyResult = sendMoney(rightStreamNode, leftStreamNode, paymentAmount);
+
+    assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
+    assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
+    assertThat(sendMoneyResult.numFulfilledPackets()).isEqualTo(1);
+    assertThat(sendMoneyResult.numRejectPackets()).isEqualTo(0);
+
+    logger.info("Payment Sent: {}", sendMoneyResult);
+  }
+
   /**
    * Simulate a left-to-right STREAM with 20% packet loss due to T03 errors.
    */
@@ -150,14 +172,7 @@ public class SenderReceiverTest {
     );
     this.initIlpNetworkForStream(simulatedIlpNetwork);
 
-    final StreamConnectionDetails connectionDetails = leftStreamNode.getNewStreamConnectionDetails();
-    final SendMoneyResult sendMoneyResult = leftStreamNode.streamSender().sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        LEFT_SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount
-    ).join();
-
+    final SendMoneyResult sendMoneyResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.numFulfilledPackets()).isEqualTo(100);
@@ -166,6 +181,85 @@ public class SenderReceiverTest {
     logger.info("Payment Sent: {}", sendMoneyResult);
   }
 
+  @Test
+  public void sendBothDirections() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(10000);
+    SendMoneyResult leftToRightResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
+    SendMoneyResult rightToLeftResult = sendMoney(rightStreamNode, leftStreamNode, paymentAmount);
+
+    Lists.newArrayList(leftToRightResult, rightToLeftResult).forEach(result -> {
+      assertThat(result.successfulPayment()).isTrue();
+      assertThat(result.amountDelivered()).isEqualTo(paymentAmount);
+      assertThat(result.originalAmount()).isEqualTo(paymentAmount);
+      assertThat(result.numFulfilledPackets()).isGreaterThan(1);
+      assertThat(result.numRejectPackets()).isEqualTo(0);
+    });
+  }
+
+  @Test
+  public void sendFromLeftToRightMultiThreadedSharedSender() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(10000);
+    int parallelism = 40;
+    int runCount = 100;
+
+    SimpleStreamSender sender = new SimpleStreamSender(leftStreamNode.link());
+
+    List<CompletableFuture<SendMoneyResult>> results =
+        runInParallel(parallelism, runCount, () -> sendMoney(sender, leftStreamNode, rightStreamNode, paymentAmount));
+
+    awaitResults(results).forEach(result -> {
+      assertThat(result.numFulfilledPackets()).isGreaterThan(1);
+      assertThat(result.numRejectPackets()).isEqualTo(0);
+      assertThat(result.successfulPayment()).isTrue();
+      assertThat(result.amountDelivered()).isEqualTo(paymentAmount);
+      assertThat(result.originalAmount()).isEqualTo(paymentAmount);
+    });
+  }
+
+  @Test
+  public void sendBothDirectionsMultiThreaded() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(100000);
+    int parallelism = 50;
+    int runs = 200;
+
+    // queue up left-to-right send
+    SimpleStreamSender sendLeft = new SimpleStreamSender(leftStreamNode.link());
+    List<CompletableFuture<SendMoneyResult>> results =
+        runInParallel(parallelism, runs, () -> sendMoney(sendLeft, leftStreamNode, rightStreamNode, paymentAmount));
+
+    // queue up right-to-left send
+    SimpleStreamSender sendRight = new SimpleStreamSender(leftStreamNode.link());
+    results.addAll(
+        runInParallel(parallelism, runs, () -> sendMoney(sendRight, rightStreamNode, leftStreamNode, paymentAmount)));
+
+    awaitResults(results).forEach(result -> {
+      assertThat(result.successfulPayment()).isTrue();
+      assertThat(result.amountDelivered()).isEqualTo(paymentAmount);
+      assertThat(result.originalAmount()).isEqualTo(paymentAmount);
+      assertThat(result.numFulfilledPackets()).isGreaterThan(1);
+      assertThat(result.numRejectPackets()).isEqualTo(0);
+    });
+  }
+
+  private SendMoneyResult sendMoney(StreamNode fromNode, StreamNode toNode, UnsignedLong paymentAmount) {
+    return sendMoney(new SimpleStreamSender(fromNode.link()), fromNode, toNode, paymentAmount);
+  }
+
+  private SendMoneyResult sendMoney(StreamSender sender, StreamNode fromNode, StreamNode toNode, UnsignedLong paymentAmount) {
+    final StreamConnectionDetails connectionDetails = toNode.getNewStreamConnectionDetails();
+    return sender.sendMoney(
+        SendMoneyRequest.builder()
+            .sharedSecret(connectionDetails.sharedSecret())
+            .amount(paymentAmount)
+            .denomination(fromNode.denomination())
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sourceAddress(fromNode.senderAddress())
+            .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+            .timeout(Duration.ofMillis(10000))
+            .build())
+        .join();
+  }
+
   /////////////////
   // Helper Methods
   /////////////////
@@ -179,13 +273,7 @@ public class SenderReceiverTest {
         SimulatedPathConditions.builder().build()
     ));
 
-    final StreamConnectionDetails connectionDetails = leftStreamNode.getNewStreamConnectionDetails();
-    final SendMoneyResult sendMoneyResult = leftStreamNode.streamSender().sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        LEFT_SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount
-    ).join();
+    final SendMoneyResult sendMoneyResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
@@ -222,37 +310,21 @@ public class SenderReceiverTest {
   }
 
   private StreamNode initLeftNode() {
-    final byte[] serverSecret = BaseEncoding.base16().decode(SHARED_SECRET_HEX);
-    final StreamEncryptionService streamEncryptionService = new JavaxStreamEncryptionService();
-
-    SimpleStreamSender streamSender = new SimpleStreamSender(
-        streamEncryptionService, simulatedIlpNetwork.getLeftToRightLink()
-    );
-
-    StatelessStreamReceiver streamReceiver = new StatelessStreamReceiver(
-        () -> serverSecret,
-        new SpspStreamConnectionGenerator(),
-        streamEncryptionService,
-        StreamCodecContextFactory.oer()
-    );
-
-    return StreamNode.builder()
-        .serverSecret(serverSecret)
-        .senderAddress(LEFT_SENDER_ADDRESS)
-        .receiverAddress(RIGHT_RECEIVER_ADDRESS)
-        .streamSender(streamSender)
-        .streamReceiver(streamReceiver)
-        .denomination(Denomination.builder().assetCode("XRP").assetScale((short) 6).build())
-        .link(simulatedIlpNetwork.getLeftToRightLink())
-        .build();
+    return initNode(simulatedIlpNetwork.getLeftToRightLink(), LEFT_SENDER_ADDRESS, RIGHT_RECEIVER_ADDRESS);
   }
 
   private StreamNode initRightNode() {
+    return initNode(simulatedIlpNetwork.getRightToLeftLink(), RIGHT_SENDER_ADDRESS, LEFT_RECEIVER_ADDRESS);
+  }
+
+  private static StreamNode initNode(Link link,
+                                     InterledgerAddress senderAddress,
+                                     InterledgerAddress receiverAddress) {
     final byte[] serverSecret = BaseEncoding.base16().decode(SHARED_SECRET_HEX);
     final StreamEncryptionService streamEncryptionService = new JavaxStreamEncryptionService();
 
     SimpleStreamSender streamSender = new SimpleStreamSender(
-        streamEncryptionService, simulatedIlpNetwork.getRightToLeftLink()
+        streamEncryptionService, link
     );
 
     StatelessStreamReceiver streamReceiver = new StatelessStreamReceiver(
@@ -264,15 +336,39 @@ public class SenderReceiverTest {
 
     return StreamNode.builder()
         .serverSecret(serverSecret)
-        .senderAddress(RIGHT_SENDER_ADDRESS)
-        .receiverAddress(RIGHT_RECEIVER_ADDRESS)
+        .senderAddress(senderAddress)
+        .receiverAddress(receiverAddress)
         .streamSender(streamSender)
         .streamReceiver(streamReceiver)
         .denomination(Denomination.builder().assetCode("XRP").assetScale((short) 6).build())
-        .link(simulatedIlpNetwork.getRightToLeftLink())
+        .link(link)
         .build();
   }
 
+  private <T> List<CompletableFuture<T>> runInParallel(int parallelism, int runCount, Callable<T> task) {
+    ExecutorService executorService = Executors.newFixedThreadPool(parallelism);
+    List<CompletableFuture<T>> tasks = IntStream.range(0, runCount)
+        .mapToObj((taskId) -> CompletableFuture.supplyAsync(() -> {
+          logger.info("Starting task " + taskId);
+          try {
+            T result = task.call();
+            logger.info("Finished task " + taskId);
+            return result;
+          } catch (Exception e) {
+            logger.warn("Failed task " + taskId, e);
+            throw new RuntimeException(e);
+          }
+        }, executorService))
+        .collect(Collectors.toList());
+    executorService.shutdown();
+    return tasks;
+  }
+
+  private static <T> List<T> awaitResults(List<CompletableFuture<T>> futures) {
+    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+    return futures.stream().map(CompletableFuture::join).collect(Collectors.toList());
+  }
+
   /**
    * An internal class that encapsulates all the details needed for a particular STREAM node (regardless of if its a
    * sender or receiver) for purposes of this test harness.