Repository: hyperledger/quilt
Pull request #327 (2019-10-09 19:48:42)

Nh/payment tracker modes

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

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

```diff
@@             Coverage Diff             @@
##             master    #327      +/-   ##
===========================================
- Coverage     81.01%   80.1%   -0.91%     
- Complexity      772     773       +1     
===========================================
  Files           228     233       +5     
  Lines          3577    3660      +83     
  Branches        219     222       +3     
===========================================
+ Hits           2898    2932      +34     
- Misses          584     635      +51     
+ Partials         95      93       -2
```


| [Impacted Files](https://codecov.io/gh/hyperledger/quilt/pull/327?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...ger/stream/calculators/ExchangeRateCalculator.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vY2FsY3VsYXRvcnMvRXhjaGFuZ2VSYXRlQ2FsY3VsYXRvci5qYXZh) | `100% <ø> (ø)` | `1 <0> (ø)` | :arrow_down: |
| [.../java/org/interledger/stream/SendMoneyRequest.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vU2VuZE1vbmV5UmVxdWVzdC5qYXZh) | `0% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: |
| [...n/java/org/interledger/stream/SendMoneyResult.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vU2VuZE1vbmV5UmVzdWx0LmphdmE=) | `100% <ø> (ø)` | `2 <0> (-1)` | :arrow_down: |
| [...ream/sender/FixedReceiverAmountPaymentTracker.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL3N0cmVhbS9zZW5kZXIvRml4ZWRSZWNlaXZlckFtb3VudFBheW1lbnRUcmFja2VyLmphdmE=) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...in/java/org/interledger/stream/PaymentTracker.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vUGF5bWVudFRyYWNrZXIuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...java/org/interledger/stream/UnsignedLongUtils.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vVW5zaWduZWRMb25nVXRpbHMuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...stream/calculators/NoOpExchangeRateCalculator.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vY2FsY3VsYXRvcnMvTm9PcEV4Y2hhbmdlUmF0ZUNhbGN1bGF0b3IuamF2YQ==) | `66.66% <0%> (-33.34%)` | `2 <0> (ø)` | |
| [...in/java/org/interledger/stream/PrepareAmounts.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vUHJlcGFyZUFtb3VudHMuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...stream/sender/FixedSenderAmountPaymentTracker.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL3N0cmVhbS9zZW5kZXIvRml4ZWRTZW5kZXJBbW91bnRQYXltZW50VHJhY2tlci5qYXZh) | `100% <100%> (ø)` | `0 <0> (?)` | |
| [...edger/stream/receiver/StatelessStreamReceiver.java](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tcmVjZWl2ZXIvc3JjL21haW4vamF2YS9vcmcvaW50ZXJsZWRnZXIvc3RyZWFtL3JlY2VpdmVyL1N0YXRlbGVzc1N0cmVhbVJlY2VpdmVyLmphdmE=) | `94.73% <50%> (+3.15%)` | `14 <0> (+2)` | :arrow_up: |
| ... and [6 more](https://codecov.io/gh/hyperledger/quilt/pull/327/diff?src=pr&el=tree-more) | |

------

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


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 2ef00f3..11df92c 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
@@ -1,8 +1,10 @@
 package org.interledger.examples;
 
 import static okhttp3.CookieJar.NO_COOKIES;
+
 import org.interledger.codecs.ilp.InterledgerCodecContextFactory;
 import org.interledger.core.InterledgerAddress;
+import org.interledger.core.SharedSecret;
 import org.interledger.link.Link;
 import org.interledger.link.http.IlpOverHttpLink;
 import org.interledger.link.http.IlpOverHttpLinkSettings;
@@ -13,13 +15,11 @@ 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.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
-import org.interledger.core.SharedSecret;
+import org.interledger.stream.SenderAmountMode;
 import org.interledger.stream.sender.SimpleStreamSender;
-import java.time.Duration;
-import java.util.Arrays;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.primitives.UnsignedLong;
 import okhttp3.ConnectionPool;
@@ -27,6 +27,11 @@ import okhttp3.ConnectionSpec;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
 /**
  * Example how to use Quilt to send a STREAM payment. See this module's README for more details.
  */
@@ -44,7 +49,7 @@ public class SendMoneyExample {
   private static final InterledgerAddress SENDER_ADDRESS =
       InterledgerAddress.of("test.xpring-dev.rs3").with(SENDER_ACCOUNT_USERNAME);
 
-  public static void main(String args[]) throws ExecutionException, InterruptedException {
+  public static void main(String[] args) throws ExecutionException, InterruptedException {
     // Create SPSP client
     InterledgerRustNodeClient spspClient =
         new InterledgerRustNodeClient(newHttpClient(), SENDER_ACCOUNT_USERNAME + ":" + SENDER_PASS_KEY, TESTNET_URI);
@@ -62,10 +67,17 @@ public class SendMoneyExample {
     System.out.println("Starting balance for sender: " + spspClient.getBalance(SENDER_ACCOUNT_USERNAME));
 
     // Send payment using STREAM
-    SendMoneyResult result = simpleStreamSender.sendMoney(SharedSecret.of(connectionDetails.sharedSecret().value()),
-        SENDER_ADDRESS, connectionDetails.destinationAddress(), UnsignedLong.valueOf(100000), Denominations.XRP,
-        Duration.ofMillis(30000))
-        .get();
+    SendMoneyResult result = simpleStreamSender.sendMoney(
+        SendMoneyRequest.builder()
+            .sourceAddress(SENDER_ADDRESS)
+            .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+            .amount(UnsignedLong.valueOf(100000))
+            .denomination(Denominations.XRP)
+            .destinationAddress(connectionDetails.destinationAddress())
+            .timeout(Duration.ofMillis(30000))
+            .sharedSecret(SharedSecret.of(connectionDetails.sharedSecret().value()))
+            .build()
+    ).get();
 
     System.out.println("Send money result: " + result);
     System.out.println("Ending balance for sender: " + spspClient.getBalance(SENDER_ACCOUNT_USERNAME));
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/AimdCongestionController.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/AimdCongestionController.java
index 3c3ad60..80d8dae 100644
--- a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/AimdCongestionController.java
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/AimdCongestionController.java
@@ -1,6 +1,7 @@
 package org.interledger.stream.sender;
 
 import static org.interledger.core.InterledgerErrorCode.T04_INSUFFICIENT_LIQUIDITY_CODE;
+import static org.interledger.stream.FluentCompareTo.is;
 
 import org.interledger.codecs.stream.StreamCodecContextFactory;
 import org.interledger.core.InterledgerErrorCode;
@@ -127,14 +128,14 @@ public class AimdCongestionController implements CongestionController {
     // Once we start getting errors, switch to Additive Increase, Multiplicative Decrease (AIMD) congestion avoidance.
     if (this.congestionState.get() == CongestionState.SLOW_START) {
       // Double the max in flight but don't exceed the u64 max value
-      if (HALF_UNSIGNED_MAX.compareTo(this.maxInFlight.get()) >= 0) {
+      if (is(HALF_UNSIGNED_MAX).greaterThanEqualTo(this.maxInFlight.get())) {
         this.maxInFlight.getAndUpdate(maxInFlight -> maxInFlight.times(TWO));
       } else {
         this.maxInFlight.set(UnsignedLong.MAX_VALUE);
       }
     } else {
       // Add to the max in flight but don't exceed the u64 max value
-      if (UnsignedLong.MAX_VALUE.minus(increaseAmount).compareTo(this.maxInFlight.get()) >= 0) {
+      if (is(UnsignedLong.MAX_VALUE.minus(increaseAmount)).greaterThanEqualTo(this.maxInFlight.get())) {
         this.maxInFlight.getAndUpdate(maxInFlight -> maxInFlight.plus(increaseAmount));
       } else {
         this.maxInFlight.set(UnsignedLong.MAX_VALUE);
@@ -314,7 +315,7 @@ public class AimdCongestionController implements CongestionController {
 
   @Override
   public boolean hasInFlight() {
-    return this.amountInFlight.get().compareTo(UnsignedLong.ZERO) > 0;
+    return is(this.amountInFlight.get()).greaterThan(UnsignedLong.ZERO);
   }
 
   /**
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTracker.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTracker.java
new file mode 100644
index 0000000..cdee24b
--- /dev/null
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTracker.java
@@ -0,0 +1,142 @@
+package org.interledger.stream.sender;
+
+import static org.interledger.stream.FluentCompareTo.is;
+
+import org.interledger.stream.Denomination;
+import org.interledger.stream.PrepareAmounts;
+import org.interledger.stream.ReceiverAmountPaymentTracker;
+import org.interledger.stream.SenderAmountMode;
+import org.interledger.stream.StreamUtils;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
+
+import com.google.common.primitives.UnsignedLong;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class FixedReceiverAmountPaymentTracker implements ReceiverAmountPaymentTracker {
+
+  // The original amount, in receiver's units, to send
+  private final UnsignedLong amountToDeliver;
+  // The amount, in sender's units, left to send (i.e., the unsent amount). On rare error occasions, this value
+  // _could_ diverge from 1-sentAmount.
+  private final AtomicReference<UnsignedLong> amountLeftToDeliver;
+  // The amount, in sender's units, that was sent.
+  private final AtomicReference<UnsignedLong> sentAmount;
+  // The amount, in receiver's units, that was actually delivered to the receiver.
+  private final AtomicReference<UnsignedLong> deliveredAmount;
+
+  private final ExchangeRateCalculator rateCalculator;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param amountToDeliver An {@link UnsignedLong} representing the amount to send to the receiver, denominated in the
+   *                        receiver's units.
+   */
+  public FixedReceiverAmountPaymentTracker(final UnsignedLong amountToDeliver) {
+    this(amountToDeliver, new NoOpExchangeRateCalculator());
+  }
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param amountToDeliver An {@link UnsignedLong} representing the amount to send, in the receiver's units.
+   * @param rateCalculator  An {@link ExchangeRateCalculator} that informs this tracker which amounts to use is
+   *                        subsequent operations, depending on market exchange rates, observed path rates, and possibly
+   *                        other data.
+   */
+  public FixedReceiverAmountPaymentTracker(
+      final UnsignedLong amountToDeliver, final ExchangeRateCalculator rateCalculator
+  ) {
+    this.amountToDeliver = Objects.requireNonNull(amountToDeliver);
+    this.rateCalculator = Objects.requireNonNull(rateCalculator);
+
+    amountLeftToDeliver = new AtomicReference<>(amountToDeliver);
+    sentAmount = new AtomicReference<>(UnsignedLong.ZERO);
+    deliveredAmount = new AtomicReference<>(UnsignedLong.ZERO);
+  }
+
+  @Override
+  public UnsignedLong getOriginalAmount() {
+    return amountToDeliver;
+  }
+
+  @Override
+  public SenderAmountMode getOriginalAmountMode() {
+    return SenderAmountMode.RECEIVER_AMOUNT;
+  }
+
+  @Override
+  public UnsignedLong getOriginalAmountLeft() {
+    return amountLeftToDeliver.get();
+  }
+
+  @Override
+  public UnsignedLong getAmountSent() {
+    return sentAmount.get();
+  }
+
+  @Override
+  public UnsignedLong getDeliveredAmount() {
+    return deliveredAmount.get();
+  }
+
+  @Override
+  public PrepareAmounts getSendPacketAmounts(
+      UnsignedLong congestionLimit, Denomination sendDenomination, Optional<Denomination> receiverDenomination
+  ) {
+    if (congestionLimit.equals(UnsignedLong.ZERO) || amountLeftToDeliver.get().equals(UnsignedLong.ZERO)) {
+      return PrepareAmounts.builder().amountToSend(UnsignedLong.ZERO).minimumAmountToAccept(UnsignedLong.ZERO).build();
+    }
+    UnsignedLong amountToSendInSenderUnits =
+        rateCalculator.calculateAmountToSend(amountLeftToDeliver.get(), sendDenomination, receiverDenomination.get());
+    final UnsignedLong packetAmountToSend = StreamUtils.max(StreamUtils.min(amountToSendInSenderUnits, congestionLimit),
+        UnsignedLong.ONE);
+    UnsignedLong minAmountToAcceptInReceiverUnits =
+        rateCalculator.calculateMinAmountToAccept(packetAmountToSend, sendDenomination, receiverDenomination);
+    return PrepareAmounts.builder()
+        .minimumAmountToAccept(minAmountToAcceptInReceiverUnits)
+        .amountToSend(packetAmountToSend)
+        .build();
+  }
+
+  @Override
+  public void auth(PrepareAmounts prepareAmounts) {
+    reduceAmountLeftToDeliver(prepareAmounts.getMinimumAmountToAccept());
+  }
+
+  private void reduceAmountLeftToDeliver(UnsignedLong amountToReduce) {
+    this.amountLeftToDeliver.getAndUpdate(sourceAmount -> {
+      if (is(sourceAmount).lessThan(amountToReduce)) {
+        // we've overdelivered so just set the amountToDeliver to 0 since UnsignedLong cannot go negative
+        return UnsignedLong.ZERO;
+      } else {
+        return sourceAmount.minus(amountToReduce);
+      }
+    });
+  }
+
+  @Override
+  public void rollback(PrepareAmounts prepareAmounts, boolean packetRejected) {
+    this.amountLeftToDeliver.getAndUpdate(sourceAmount -> sourceAmount.plus(prepareAmounts.getMinimumAmountToAccept()));
+  }
+
+  @Override
+  public void commit(PrepareAmounts prepareAmounts, UnsignedLong deliveredAmount) {
+    if (is(prepareAmounts.getMinimumAmountToAccept()).lessThan(deliveredAmount)) {
+      UnsignedLong overrage = deliveredAmount.minus(prepareAmounts.getMinimumAmountToAccept());
+      reduceAmountLeftToDeliver(overrage);
+    }
+    this.deliveredAmount.getAndUpdate(currentAmount -> currentAmount.plus(deliveredAmount));
+    this.sentAmount.getAndUpdate(currentAmount -> currentAmount.plus(prepareAmounts.getAmountToSend()));
+  }
+
+  @Override
+  public boolean moreToSend() {
+    return is(deliveredAmount.get()).lessThan(this.amountToDeliver);
+  }
+
+}
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedSenderAmountPaymentTracker.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedSenderAmountPaymentTracker.java
new file mode 100644
index 0000000..8c55391
--- /dev/null
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/FixedSenderAmountPaymentTracker.java
@@ -0,0 +1,118 @@
+package org.interledger.stream.sender;
+
+import static org.interledger.stream.FluentCompareTo.is;
+
+import org.interledger.stream.Denomination;
+import org.interledger.stream.PrepareAmounts;
+import org.interledger.stream.SenderAmountPaymentTracker;
+import org.interledger.stream.StreamUtils;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.primitives.UnsignedLong;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class FixedSenderAmountPaymentTracker implements SenderAmountPaymentTracker {
+
+  // The original amount, in sender's units, to send
+  private final UnsignedLong amountToSend;
+  // The amount, in sender's units, left to send (i.e., the unsent amount). On rare error occasions, this value
+  // _could_ diverge from 1-sentAmount.
+  private final AtomicReference<UnsignedLong> amountLeftToSend;
+  // The amount, in sender's units, that was sent.
+  private final AtomicReference<UnsignedLong> sentAmount;
+  // The amount, in receiver's units, that was actually delivered to the receiver.
+  private final AtomicReference<UnsignedLong> deliveredAmount;
+
+  private final ExchangeRateCalculator rateCalculator;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param amountToSend An {@link UnsignedLong} representing the amount to send to the receiver, denominated in the
+   *                     sender's units.
+   */
+  public FixedSenderAmountPaymentTracker(final UnsignedLong amountToSend) {
+    this(amountToSend, new NoOpExchangeRateCalculator());
+  }
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param amountToSend   An {@link UnsignedLong} representing the amount to send, in the sender's units.
+   * @param rateCalculator An {@link ExchangeRateCalculator} that informs this tracker which amounts to use is
+   *                       subsequent operations, depending on market exchange rates, observed path rates, and possibly
+   *                       other data.
+   */
+  public FixedSenderAmountPaymentTracker(final UnsignedLong amountToSend, final ExchangeRateCalculator rateCalculator) {
+    this.amountToSend = Objects.requireNonNull(amountToSend);
+    this.rateCalculator = Objects.requireNonNull(rateCalculator);
+
+    amountLeftToSend = new AtomicReference<>(amountToSend);
+    sentAmount = new AtomicReference<>(UnsignedLong.ZERO);
+    deliveredAmount = new AtomicReference<>(UnsignedLong.ZERO);
+  }
+
+  @Override
+  public UnsignedLong getOriginalAmount() {
+    return amountToSend;
+  }
+
+  @Override
+  public UnsignedLong getOriginalAmountLeft() {
+    return amountLeftToSend.get();
+  }
+
+  @Override
+  public UnsignedLong getAmountSent() {
+    return sentAmount.get();
+  }
+
+  @Override
+  public UnsignedLong getDeliveredAmount() {
+    return deliveredAmount.get();
+  }
+
+  @Override
+  public PrepareAmounts getSendPacketAmounts(UnsignedLong congestionLimit,
+      Denomination sendDenomination,
+      Optional<Denomination> receiverDenomination) {
+    final UnsignedLong packetAmountToSend = StreamUtils.min(amountLeftToSend.get(), congestionLimit);
+    return PrepareAmounts.builder()
+        .minimumAmountToAccept(
+            rateCalculator.calculateMinAmountToAccept(packetAmountToSend, sendDenomination, receiverDenomination))
+        .amountToSend(packetAmountToSend)
+        .build();
+  }
+
+  @Override
+  public void auth(PrepareAmounts prepareAmounts) {
+    this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.minus(prepareAmounts.getAmountToSend()));
+  }
+
+  @Override
+  public void rollback(PrepareAmounts prepareAmounts, boolean packetRejected) {
+    this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.plus(prepareAmounts.getAmountToSend()));
+  }
+
+  @Override
+  public void commit(PrepareAmounts prepareAmounts, UnsignedLong deliveredAmount) {
+    this.deliveredAmount.getAndUpdate(currentAmount -> currentAmount.plus(deliveredAmount));
+    this.sentAmount.getAndUpdate(currentAmount -> currentAmount.plus(prepareAmounts.getAmountToSend()));
+  }
+
+  @Override
+  public boolean moreToSend() {
+    return is(this.sentAmount.get()).lessThan(this.amountToSend);
+  }
+
+  @VisibleForTesting
+  void setSentAmount(UnsignedLong sentAmount) {
+    this.sentAmount.set(sentAmount);
+  }
+
+}
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 65e553d..6edab09 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
@@ -19,15 +19,14 @@ 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.PaymentTracker;
+import org.interledger.stream.PrepareAmounts;
 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;
@@ -64,8 +63,8 @@ import java.util.concurrent.ThreadFactory;
 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;
 
 /**
@@ -127,7 +126,6 @@ public class SimpleStreamSender implements StreamSender {
     this(streamEncryptionService, link, executorService, new StreamConnectionManager());
   }
 
-
   /**
    * Required-args Constructor.
    *
@@ -162,46 +160,6 @@ public class SimpleStreamSender implements StreamSender {
     return Executors.newFixedThreadPool(30, factory);
   }
 
-  @Override
-  public CompletableFuture<SendMoneyResult> sendMoney(
-      final SharedSecret sharedSecret,
-      final InterledgerAddress sourceAddress,
-      final InterledgerAddress destinationAddress,
-      final UnsignedLong amount,
-      final Denomination denomination
-  ) {
-    SendMoneyRequest request = SendMoneyRequest.builder()
-        .sharedSecret(sharedSecret)
-        .destinationAddress(destinationAddress)
-        .sourceAddress(sourceAddress)
-        .amount(amount)
-        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
-        .denomination(denomination)
-        .build();
-    return sendMoney(request);
-  }
-
-  @Override
-  public CompletableFuture<SendMoneyResult> sendMoney(
-      final SharedSecret sharedSecret,
-      final InterledgerAddress sourceAddress,
-      final InterledgerAddress destinationAddress,
-      final UnsignedLong amount,
-      final Denomination denomination,
-      final Duration timeout
-  ) {
-    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);
@@ -280,38 +238,29 @@ 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;
-    // The original amount, in sender's units, to send
-    private final AtomicReference<UnsignedLong> originalAmountToSend;
-    // The amount, in sender's units, left to send (i.e., the unsent amount). On rare error occasions, this value
-    // _could_ diverge from 1-sentAmount.
-    private final AtomicReference<UnsignedLong> amountLeftToSend;
-    // The amount, in sender's units, that was sent.
-    private final AtomicReference<UnsignedLong> sentAmount;
-    // The amount, in receiver's units, that was actually delivered to the receiver.
-    private final AtomicReference<UnsignedLong> deliveredAmount;
+
     private final AtomicBoolean shouldSendSourceAddress;
     private final AtomicInteger numFulfilledPackets;
     private final AtomicInteger numRejectedPackets;
-
+    private final PaymentTracker paymentTracker;
     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 request                      all relevant details about the money to send
+     * @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,
@@ -335,18 +284,14 @@ public class SimpleStreamSender implements StreamSender {
       this.sourceAddress = request.sourceAddress();
       this.destinationAddress = request.destinationAddress();
 
-      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 = request.timeout();
 
-      this.exchangeRateCalculator = request.exchangeRateCalculator();
       this.senderDenomination = request.denomination();
+
+      this.paymentTracker = request.paymentTracker();
     }
 
     /**
@@ -357,7 +302,6 @@ public class SimpleStreamSender implements StreamSender {
     CompletableFuture<SendMoneyResult> send() {
       Objects.requireNonNull(sharedSecret);
       Objects.requireNonNull(destinationAddress);
-      Objects.requireNonNull(originalAmountToSend);
 
       Instant startPreflight = Instant.now();
       try {
@@ -369,8 +313,9 @@ public class SimpleStreamSender implements StreamSender {
             .numFulfilledPackets(0)
             .amountDelivered(UnsignedLong.ZERO)
             .amountSent(UnsignedLong.ZERO)
-            .originalAmount(originalAmountToSend.get())
-            .amountLeftToSend(originalAmountToSend.get())
+            .originalAmount(paymentTracker.getOriginalAmount())
+            .amountLeftToSend(paymentTracker.getOriginalAmountLeft())
+            .successfulPayment(paymentTracker.successful())
             .build());
       } catch (Exception e) {
         logger.warn("Preflight check failed", e);
@@ -387,13 +332,14 @@ 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())
+                .amountDelivered(paymentTracker.getDeliveredAmount())
+                .amountSent(paymentTracker.getAmountSent())
+                .amountLeftToSend(paymentTracker.getOriginalAmountLeft())
+                .originalAmount(paymentTracker.getOriginalAmount())
                 .numFulfilledPackets(numFulfilledPackets.get())
                 .numRejectPackets(numRejectedPackets.get())
                 .sendMoneyDuration(Duration.between(start, Instant.now()))
+                .successfulPayment(paymentTracker.successful())
                 .build();
           }, sendMoneyExecutor)
           .whenComplete(($, error) -> {
@@ -431,14 +377,15 @@ public class SimpleStreamSender implements StreamSender {
               // This aggregator supports only a simple stream-id, which is one.
               .streamId(UnsignedLong.ONE)
               .shares(UnsignedLong.ONE)
+              .build(),
+          ConnectionNewAddressFrame.builder()
+              .sourceAddress(sourceAddress)
+              .build(),
+          ConnectionAssetDetailsFrame.builder()
+              .sourceDenomination(senderDenomination)
               .build()
       );
 
-      frames.add(ConnectionNewAddressFrame.builder()
-          .sourceAddress(sourceAddress)
-          .build()
-      );
-
       final StreamPacket streamPacket = StreamPacket.builder()
           .interledgerPacketType(InterledgerPacketType.PREPARE)
           .prepareAmount(UnsignedLong.ZERO)
@@ -492,7 +439,12 @@ public class SimpleStreamSender implements StreamSender {
 
       while (soldierOn(timeoutReached.get())) {
         // Determine the amount to send
-        final UnsignedLong amountToSend = StreamUtils.min(amountLeftToSend.get(), congestionController.getMaxAmount());
+        PrepareAmounts amounts = paymentTracker.getSendPacketAmounts(congestionController.getMaxAmount(),
+            senderDenomination,
+            receiverDenomination);
+        UnsignedLong amountToSend = amounts.getAmountToSend();
+        UnsignedLong receiverMinimum = amounts.getMinimumAmountToAccept();
+
         if (amountToSend.equals(UnsignedLong.ZERO) || timeoutReached.get()) {
           try {
             // Don't send any more, but wait a bit for outstanding requests to complete so we don't cycle needlessly in
@@ -525,9 +477,6 @@ public class SimpleStreamSender implements StreamSender {
                 .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.
@@ -549,7 +498,12 @@ public class SimpleStreamSender implements StreamSender {
             .data(streamPacketData)
             .build();
 
-        this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.minus(amountToSend));
+        // auth
+        // capture
+        // rollback
+
+        final PrepareAmounts prepareAmounts = PrepareAmounts.from(preparePacket, streamPacket);
+        paymentTracker.auth(prepareAmounts);
 
         try {
           // don't submit new tasks if the timeout was reached within this iteration of the while loop
@@ -559,14 +513,14 @@ 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);
-            schedule(timeoutReached, preparePacket, streamPacket, amountToSend);
+            schedule(timeoutReached, preparePacket, streamPacket, prepareAmounts);
           } 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...
-          this.amountLeftToSend.getAndUpdate(sourceAmount -> sourceAmount.plus(amountToSend));
+          paymentTracker.rollback(prepareAmounts, false);
           logger.error("Submit failed", e);
         }
       }
@@ -575,33 +529,33 @@ public class SimpleStreamSender implements StreamSender {
 
     @VisibleForTesting
     void schedule(AtomicBoolean timeoutReached, InterledgerPreparePacket preparePacket, StreamPacket streamPacket,
-                  UnsignedLong amountToSend) {
+                  PrepareAmounts prepareAmounts) {
       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)
+                  fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket, prepareAmounts),
+                  rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, prepareAmounts,
+                      numRejectedPackets, congestionController)
               );
             } catch (Exception e) {
               logger.error("Link send failed. preparePacket={}", preparePacket, e);
-              congestionController.reject(amountToSend, InterledgerRejectPacket.builder()
+              congestionController.reject(preparePacket.getAmount(), 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));
+              paymentTracker.rollback(prepareAmounts, false);
             }
           }
         });
       } 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()
+        congestionController.reject(preparePacket.getAmount(), InterledgerRejectPacket.builder()
             .code(InterledgerErrorCode.F00_BAD_REQUEST)
             .message(
                 String.format("Unable to schedule sendMoney task. preparePacket=%s error=%s", preparePacket,
@@ -620,7 +574,7 @@ public class SimpleStreamSender implements StreamSender {
       //   and you haven't delivered the full amount
       //   and you haven't timed out
       return this.congestionController.hasInFlight()
-          || (!streamConnection.isClosed() && moreToSend() && !timeoutReached);
+          || (!streamConnection.isClosed() && paymentTracker.moreToSend() && !timeoutReached);
     }
 
     /**
@@ -678,11 +632,13 @@ public class SimpleStreamSender implements StreamSender {
     void handleFulfill(
         final InterledgerPreparePacket originalPreparePacket,
         final StreamPacket originalStreamPacket,
-        final InterledgerFulfillPacket fulfillPacket
+        final InterledgerFulfillPacket fulfillPacket,
+        final PrepareAmounts prepareAmounts
     ) {
       Objects.requireNonNull(originalPreparePacket);
       Objects.requireNonNull(originalStreamPacket);
       Objects.requireNonNull(fulfillPacket);
+      Objects.requireNonNull(prepareAmounts);
 
       this.numFulfilledPackets.getAndIncrement();
 
@@ -695,8 +651,8 @@ public class SimpleStreamSender implements StreamSender {
       //if let Ok (packet) = StreamPacket::from_encrypted ( & self.shared_secret, fulfill.into_data()){
       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(originalPreparePacket.getAmount()));
+        UnsignedLong deliveredAmount = streamPacket.prepareAmount();
+        paymentTracker.commit(prepareAmounts, deliveredAmount);
       } else {
         logger.warn("Unable to parse STREAM packet from fulfill data. "
                 + "originalPreparePacket={} originalStreamPacket={} fulfillPacket={}",
@@ -705,7 +661,7 @@ public class SimpleStreamSender implements StreamSender {
 
       logger.debug("Prepare packet fulfilled ({} left to send). "
               + "originalPreparePacket={} originalStreamPacket={} fulfillPacket={}",
-          this.amountLeftToSend, originalPreparePacket, originalStreamPacket, fulfillPacket
+          paymentTracker.getOriginalAmountLeft(), originalPreparePacket, originalStreamPacket, fulfillPacket
       );
     }
 
@@ -722,20 +678,22 @@ public class SimpleStreamSender implements StreamSender {
         final InterledgerPreparePacket originalPreparePacket,
         final StreamPacket originalStreamPacket,
         final InterledgerRejectPacket rejectPacket,
+        final PrepareAmounts prepareAmounts,
         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);
+      Objects.requireNonNull(prepareAmounts);
 
       final UnsignedLong amountToSend = originalPreparePacket.getAmount();
+
       numRejectedPackets.getAndIncrement();
-      amountLeftToSend.getAndUpdate(currentAmount -> currentAmount.plus(amountToSend));
+
+      paymentTracker.rollback(prepareAmounts, true);
       congestionController.reject(amountToSend, rejectPacket);
 
       logger.debug(
@@ -743,7 +701,7 @@ public class SimpleStreamSender implements StreamSender {
               + "originalStreamPacket={} rejectPacket={}",
           amountToSend,
           rejectPacket.getCode().getCode(),
-          amountLeftToSend.get(),
+          paymentTracker.getOriginalAmountLeft(),
           originalPreparePacket,
           originalStreamPacket,
           rejectPacket
@@ -834,9 +792,11 @@ public class SimpleStreamSender implements StreamSender {
           .data(encryptedStreamPacket)
           .build();
 
+      final PrepareAmounts prepareAmounts = PrepareAmounts.from(preparePacket, streamPacket);
+
       link.sendPacket(preparePacket).handle(
-          fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket),
-          rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, numRejectedPackets, amountLeftToSend,
+          fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket, prepareAmounts),
+          rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, prepareAmounts, numRejectedPackets,
               congestionController)
       );
 
@@ -857,7 +817,8 @@ public class SimpleStreamSender implements StreamSender {
           .ifPresent($ -> {
             logger.info(
                 "StreamId {} Closed. Delivered: {} ({} packets fulfilled, {} packets rejected)",
-                $.streamId(), this.deliveredAmount, this.numFulfilledPackets.get(), this.numRejectedPackets.get()
+                $.streamId(), paymentTracker.getDeliveredAmount(), this.numFulfilledPackets.get(),
+                this.numRejectedPackets.get()
             );
           });
     }
@@ -870,28 +831,13 @@ public class SimpleStreamSender implements StreamSender {
     @VisibleForTesting
     SendMoneyResult collectSendMoneyStatistics() {
       return SendMoneyResult.builder()
-          .amountDelivered(this.deliveredAmount.get())
-          .amountSent(this.sentAmount.get())
-          .amountLeftToSend(this.amountLeftToSend.get())
+          .amountDelivered(paymentTracker.getDeliveredAmount())
+          .amountSent(paymentTracker.getAmountSent())
+          .amountLeftToSend(paymentTracker.getOriginalAmountLeft())
           .numFulfilledPackets(this.numFulfilledPackets.get())
           .numRejectPackets(this.numRejectedPackets.get())
+          .successfulPayment(paymentTracker.successful())
           .build();
     }
-
-    @VisibleForTesting
-    boolean moreToSend() {
-      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 sentAmount An {@link UnsignedLong} representing the deliveredAmount to set.
-     */
-    @VisibleForTesting
-    void setAmountSentForTesting(final UnsignedLong sentAmount) {
-      Objects.requireNonNull(sentAmount);
-      this.sentAmount.set(sentAmount);
-    }
   }
 }
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 828a0d3..ae1f056 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,13 +1,8 @@
 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 java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -17,63 +12,6 @@ import java.util.concurrent.CompletableFuture;
  */
 public interface StreamSender {
 
-  /**
-   * <p>Send "money" (i.e., some unit of value) from a source ILP address to a destination address.</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 sharedSecret       A {@link SharedSecret} known only to the sender and receiver, negotiated using some
-   *                           higher-level protocol (e.g., SPSP or something else).
-   * @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.
-   */
-  CompletableFuture<SendMoneyResult> sendMoney(
-      final SharedSecret sharedSecret,
-      final InterledgerAddress sourceAddress,
-      final InterledgerAddress destinationAddress,
-      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 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
-   * supported in the future.</p>
-   *
-   * @param sharedSecret       A {@link SharedSecret} known only to the sender and receiver, negotiated using some
-   *                           higher-level protocol (e.g., SPSP or something else).
-   * @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,
-   *                           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.
-   *
-   * @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.
-   */
-  CompletableFuture<SendMoneyResult> sendMoney(
-      final SharedSecret sharedSecret,
-      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>
@@ -83,6 +21,7 @@ public interface StreamSender {
    * 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
index d850fe5..417926d 100644
--- 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
@@ -1,9 +1,8 @@
 package org.interledger.stream;
 
-import org.interledger.stream.Denomination;
-
 public final class Denominations {
-  private Denominations(){}
+  private Denominations() {
+  }
 
   public static final Denomination XRP = Denomination.builder()
       .assetCode("XRP")
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/FluentCompareTo.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/FluentCompareTo.java
new file mode 100644
index 0000000..9ebbaad
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/FluentCompareTo.java
@@ -0,0 +1,100 @@
+package org.interledger.stream;
+
+/**
+ * Provides a fluent, readable alternative to {@link Comparable#compareTo(Object)}.
+ * Instead of code like {@code someValue.compareTo(someOtherValue) <= 0} you can write
+ * {@code is(someValue).lessThanOrEqualTo(someOtherValue) }. Also provides stronger type safety so
+ * you can't accidentally compare apples to oranges.
+ */
+public class FluentCompareTo<T extends Comparable<? super T>> {
+
+  /**
+   * Create a {@link FluentCompareTo} for the given value.
+   *
+   * @param value value to wrap
+   * @param <T>   the Java type of the wrapped value
+   * @return wrapped value
+   */
+  public static <T extends Comparable<? super T>> FluentCompareTo<T> is(T value) {
+    return new FluentCompareTo<T>(value);
+  }
+
+  private final T value;
+
+  private FluentCompareTo(T value) {
+    this.value = value;
+  }
+
+  /**
+   * Checks if wrapped value is equal than the given one.
+   *
+   * @param other given value
+   * @return true if wrapped value is equal to given value
+   */
+  public boolean equalTo(T other) {
+    return value.equals(other);
+  }
+
+  /**
+   * Checks if wrapped value less than than the given one.
+   *
+   * @param other given value
+   * @return true if wrapped value is less than given value
+   */
+  public boolean lessThan(T other) {
+    return value.compareTo(other) < 0;
+  }
+
+  /**
+   * Checks if wrapped value less than or equal to than the given one.
+   *
+   * @param other given value
+   * @return true if wrapped value is less than or equal given value
+   */
+  public boolean lessThanOrEqualTo(T other) {
+    return value.compareTo(other) <= 0;
+  }
+
+  /**
+   * Checks if wrapped value greater than than the given one.
+   *
+   * @param other given value
+   * @return true if wrapped value is greater than given value
+   */
+  public boolean greaterThan(T other) {
+    return value.compareTo(other) > 0;
+  }
+
+  /**
+   * Checks if wrapped value greater than or equal to than the given one.
+   *
+   * @param other given value
+   * @return true if wrapped value is greater than given value
+   */
+  public boolean greaterThanEqualTo(T other) {
+    return value.compareTo(other) >= 0;
+  }
+
+  /**
+   * Checks if wrapped value is between 2 given values (inclusive lower bound, exclusive upper bound).
+   *
+   * @param startInclusive lower bound inclusive value
+   * @param endExclusive   upper bound exclusive value
+   * @return true if between
+   */
+  public boolean between(T startInclusive, T endExclusive) {
+    return greaterThanEqualTo(startInclusive) && lessThan(endExclusive);
+  }
+
+  /**
+   * Checks if wrapped value is between 2 given values (exclusive lower bound, exclusive upper bound).
+   *
+   * @param startExclusive lower bound exclusive value
+   * @param endExclusive   upper bound exclusive value
+   * @return true if between
+   */
+  public boolean betweenExclusive(T startExclusive, T endExclusive) {
+    return greaterThan(startExclusive) && lessThan(endExclusive);
+  }
+
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/PaymentTracker.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/PaymentTracker.java
new file mode 100644
index 0000000..aee4900
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/PaymentTracker.java
@@ -0,0 +1,44 @@
+package org.interledger.stream;
+
+import com.google.common.primitives.UnsignedLong;
+
+import java.util.Optional;
+
+/**
+ * Defines how to track a payment while considering the amount sent vs amount received, allowing room for
+ * path-exchange-rate fluctuations and implementation-defined rules relating to whether or not to continue a payment.
+ */
+public interface PaymentTracker<T extends SenderAmountMode> {
+
+  UnsignedLong getOriginalAmount();
+
+  /**
+   * Returns the {@link SenderAmountMode} for this payment tracker.
+   *
+   * @return A {@link SenderAmountMode} that indicates the meaning of {@link #getOriginalAmount()}.
+   */
+  T getOriginalAmountMode();
+
+  UnsignedLong getOriginalAmountLeft();
+
+  UnsignedLong getAmountSent();
+
+  UnsignedLong getDeliveredAmount();
+
+  PrepareAmounts getSendPacketAmounts(UnsignedLong congestionLimit,
+      Denomination sendDenomination,
+      Optional<Denomination> receiverDenomination);
+
+  void auth(PrepareAmounts prepareAmounts);
+
+  void rollback(PrepareAmounts prepareAmounts, boolean packetRejected);
+
+  void commit(PrepareAmounts prepareAmounts, UnsignedLong deliveredAmount);
+
+  boolean moreToSend();
+
+
+  default boolean successful() {
+    return !moreToSend();
+  }
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/PrepareAmounts.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/PrepareAmounts.java
new file mode 100644
index 0000000..e57e746
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/PrepareAmounts.java
@@ -0,0 +1,51 @@
+package org.interledger.stream;
+
+import org.interledger.core.Immutable;
+import org.interledger.core.InterledgerPreparePacket;
+
+import com.google.common.primitives.UnsignedLong;
+
+/**
+ * A container object that captures amounts used in auth, rollback, and commit operations inside of the Stream Sender.
+ */
+@Immutable
+public interface PrepareAmounts {
+
+  static PrepareAmountsBuilder builder() {
+    return new PrepareAmountsBuilder();
+  }
+
+  /**
+   * Helper method to construct an instance of {@link PrepareAmounts} from a Prepare packet and an embedded Stream
+   * Packet.
+   *
+   * @param preparePacket An {@link InterledgerPreparePacket} used in a Stream send operation.
+   * @param streamPacket  A {@link StreamPacket} that was enclosed in {@code preparePacket}.
+   *
+   * @return A {@link PrepareAmounts} constructed from the supplied inputs.
+   */
+  static PrepareAmounts from(final InterledgerPreparePacket preparePacket, final StreamPacket streamPacket) {
+    return PrepareAmounts.builder()
+        .amountToSend(preparePacket.getAmount())
+        .minimumAmountToAccept(streamPacket.prepareAmount())
+        .build();
+  }
+
+  /**
+   * The amount that should be sent in a prepare packet for purposes of STREAM.
+   *
+   * @return An {@link UnsignedLong} that represents the `amount` value in an ILPv4 Prepare packet when used by a STREAM
+   *     sender.
+   */
+  UnsignedLong getAmountToSend();
+
+  /**
+   * The amount that should be put into a Stream Packet's amount field to indicate the minimum amount a receiver should
+   * accept for a given Stream Packet payment.
+   *
+   * @return An {@link UnsignedLong} that represents the minimum `amount` that a Stream Receiver should accept in an
+   *     individual Stream Packet.
+   */
+  UnsignedLong getMinimumAmountToAccept();
+
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/ReceiverAmountPaymentTracker.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/ReceiverAmountPaymentTracker.java
new file mode 100644
index 0000000..966af7b
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/ReceiverAmountPaymentTracker.java
@@ -0,0 +1,13 @@
+package org.interledger.stream;
+
+/**
+ * An extension of {@link PaymentTracker} that defines the {@link #getOriginalAmount()} to be in the receiver's units.
+ */
+public interface ReceiverAmountPaymentTracker extends PaymentTracker<SenderAmountMode> {
+
+  @Override
+  default SenderAmountMode getOriginalAmountMode() {
+    return SenderAmountMode.RECEIVER_AMOUNT;
+  }
+
+}
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
index 6b411e2..32920ae 100644
--- 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
@@ -1,10 +1,11 @@
 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 com.google.common.primitives.UnsignedLong;
+import org.immutables.value.Value.Check;
 
 import java.time.Duration;
 import java.util.Optional;
@@ -12,54 +13,83 @@ import java.util.Optional;
 @Immutable
 public interface SendMoneyRequest {
 
-  static SendMoneyRequestBuilder builder() { return new SendMoneyRequestBuilder(); }
+  static SendMoneyRequestBuilder builder() {
+    return new SendMoneyRequestBuilder();
+  }
 
   /**
+   * The shared secret, shared between only the sender and receiverd, required by IL-RFC-29 to encrypt stream frames so
+   * that only the sender and receiver can decrypt them.
    *
-   * @return A {@link SharedSecret} known only to the sender and receiver, negotiated using
-   *         some higher-level protocol (e.g., SPSP or something else).
+   * @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();
 
   /**
+   * The source address of this request.
    *
    * @return The {@link InterledgerAddress} of the source of this payment.
    */
   InterledgerAddress sourceAddress();
 
   /**
+   * The destination address of this request.
    *
    * @return The {@link InterledgerAddress} of the receiver of this payment.
    */
   InterledgerAddress destinationAddress();
 
   /**
+   * The amount of units to send. The denomination of this unit depends on the value of {@link #getSenderAmountMode()}.
    *
    * @return An {@link UnsignedLong} containing the amount of this payment.
    */
   UnsignedLong amount();
 
   /**
+   * Returns the {@link SenderAmountMode} for this payment tracker.
    *
-   * @return
+   * @return A {@link SenderAmountMode} that indicates the meaning of {@link #amount()}.
+   */
+  SenderAmountMode getSenderAmountMode();
+
+  /**
+   * The {@link Denomination} of the amount in this request.
+   *
+   * @return A {@link Denomination}.
    */
   Denomination denomination();
 
   /**
+   * <p>A {@link Duration} to wait before no longer scheduling any more requests to send stream packets for this
+   * payment.</p>
+   *
+   * <p>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.</p>
    *
-   * @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.
+   * @return A {@link Duration} of time to wait before terminating this payment.
    */
   Optional<Duration> timeout();
 
   /**
+   * Payment tracker that will monitor payment packets.
    *
-   * @return a calculator to figure out how much money should arrive on the other side of a given payment.
+   * @return payment tracker
    */
-  ExchangeRateCalculator exchangeRateCalculator();
+  PaymentTracker<SenderAmountMode> paymentTracker();
 
+  /**
+   * Ensures that this request's SenderAmountMode is compatible with {@link PaymentTracker#getOriginalAmountMode()}.
+   *
+   * @return {@code true} if the SenderAmountMode is compatible with the PaymentTracker; {@code false} otherwise.
+   */
+  @Check
+  default SendMoneyRequest check() {
+    if (this.paymentTracker().getOriginalAmountMode() != this.getSenderAmountMode()) {
+      throw new IllegalStateException("PaymentTracker is not compatible with this SenderAmountMode");
+    }
+    return this;
+  }
 }
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 3c09104..cef9aaa 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
@@ -82,12 +82,10 @@ public interface SendMoneyResult {
   Duration sendMoneyDuration();
 
   /**
-   * Compare the amount delivered to the original amount to see if they match.
+   * Indicates if the payment was completed successfully.
    *
-   * @return true if delivered matches original.
+   * @return true if payment was successful
    */
-  @Value.Derived
-  default boolean successfulPayment() {
-    return amountSent().equals(originalAmount());
-  }
+  boolean successfulPayment();
+
 }
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountMode.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountMode.java
new file mode 100644
index 0000000..ac96bcf
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountMode.java
@@ -0,0 +1,17 @@
+package org.interledger.stream;
+
+/**
+ * Defines the meaning of the {@link SendMoneyRequest#amount()} property.
+ */
+public enum SenderAmountMode {
+
+  /**
+   * The amount to send is denominated in the sender's units.
+   */
+  SENDER_AMOUNT,
+
+  /**
+   * The amount to send is denominated in the receiver's units.
+   */
+  RECEIVER_AMOUNT
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountPaymentTracker.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountPaymentTracker.java
new file mode 100644
index 0000000..d12f035
--- /dev/null
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/SenderAmountPaymentTracker.java
@@ -0,0 +1,13 @@
+package org.interledger.stream;
+
+/**
+ * An extension of {@link PaymentTracker} that defines the {@link #getOriginalAmount()} to be in the sender's units.
+ */
+public interface SenderAmountPaymentTracker extends PaymentTracker<SenderAmountMode> {
+
+  @Override
+  default SenderAmountMode getOriginalAmountMode() {
+    return SenderAmountMode.SENDER_AMOUNT;
+  }
+
+}
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamConnection.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamConnection.java
index 54f95ac..9dfbac6 100644
--- a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamConnection.java
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamConnection.java
@@ -1,5 +1,7 @@
 package org.interledger.stream;
 
+import static org.interledger.stream.FluentCompareTo.is;
+
 import org.interledger.core.InterledgerAddress;
 import org.interledger.core.SharedSecret;
 
@@ -94,7 +96,7 @@ public class StreamConnection implements Closeable {
   boolean sequenceIsSafeForSingleSharedSecret(final UnsignedLong sequence) {
     // Only return true if the Connection is not closed, and the `sequence` is below MAX_FRAMES_PER_CONNECTION (above
     // that value is unsafe).
-    return !isClosed() && Objects.requireNonNull(sequence).compareTo(MAX_FRAMES_PER_CONNECTION) <= 0;
+    return !isClosed() && is(Objects.requireNonNull(sequence)).lessThanOrEqualTo(MAX_FRAMES_PER_CONNECTION);
   }
 
   public Instant getCreationDateTime() {
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamPacket.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamPacket.java
index dcd9754..97225da 100644
--- a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamPacket.java
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamPacket.java
@@ -20,6 +20,8 @@ package org.interledger.stream;
  * =========================LICENSE_END==================================
  */
 
+import static org.interledger.stream.FluentCompareTo.is;
+
 import org.interledger.core.Immutable;
 import org.interledger.core.InterledgerPacketType;
 import org.interledger.stream.frames.StreamFrame;
@@ -113,7 +115,7 @@ public interface StreamPacket {
   @Derived
   default boolean sequenceIsSafeForSingleSharedSecret() {
     // Only return true if the `sequence` is below MAX_FRAMES_PER_CONNECTION. At or above is unsafe.
-    return this.sequence().compareTo(StreamConnection.MAX_FRAMES_PER_CONNECTION) < 0;
+    return is(this.sequence()).lessThan(StreamConnection.MAX_FRAMES_PER_CONNECTION);
   }
 
   /**
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamUtils.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamUtils.java
index 2dba83f..3545594 100644
--- a/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamUtils.java
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/StreamUtils.java
@@ -1,5 +1,7 @@
 package org.interledger.stream;
 
+import static org.interledger.stream.FluentCompareTo.is;
+
 import org.interledger.core.InterledgerCondition;
 import org.interledger.core.InterledgerFulfillment;
 import org.interledger.core.SharedSecret;
@@ -77,7 +79,7 @@ public class StreamUtils {
     Objects.requireNonNull(value1);
     Objects.requireNonNull(value2);
 
-    if (value1.compareTo(value2) < 0) {
+    if (is(value1).lessThan(value2)) {
       return value1;
     } else {
       return value2;
@@ -96,7 +98,7 @@ public class StreamUtils {
     Objects.requireNonNull(value1);
     Objects.requireNonNull(value2);
 
-    if (value1.compareTo(value2) > 0) {
+    if (is(value1).greaterThan(value2)) {
       return value1;
     } else {
       return value2;
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
index 009ca19..8998a4a 100644
--- 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
@@ -1,32 +1,65 @@
 package org.interledger.stream.calculators;
 
-import com.google.common.primitives.UnsignedLong;
 import org.interledger.stream.Denomination;
 
+import com.google.common.primitives.UnsignedLong;
+
 import java.math.BigDecimal;
 import java.util.Optional;
 
+/**
+ * <p>Defines how a STREAM sender or receiver can calculate the amount to send on a per-packet basis, in addition to
+ * acceptable amounts the receiver should accept, based upon implementation-defined market and path exchange-rate
+ * mechanisms.</p>
+ *
+ * <p>For example, an advanced implementation of this interface might decide to compare actual market exchange rates
+ * from an Oracle (e.g., the ECB) against the observed path-exchange rates. Using these metrics, the implementation
+ * could adjust amounts in an effort to make sure the proper amount of value is traversing the payment path from sender
+ * to receiver.</p>
+ */
 public interface ExchangeRateCalculator {
 
+  /**
+   * Calculates the amount to send for the ILP packet based on a desired amount to be received by the receiver (in
+   * receiver's units).
+   *
+   * @param amountToReceive     amount to be received (in receiver's units)
+   * @param sendDenomination    sender's denomination
+   * @param receiveDenomination receiver's denomination
+   *
+   * @return amount to send in the ILP packet (in sender's denomination)
+   *
+   * @throws NoExchangeRateException if exchange rate could not be calculated
+   */
+  UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive,
+      Denomination sendDenomination,
+      Denomination receiveDenomination)
+      throws NoExchangeRateException;
+
   /**
    * 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.
+   * @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;
+  UnsignedLong calculateMinAmountToAccept(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
+   * 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) {
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
index cdfa9ab..432aa8b 100644
--- 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
@@ -1,14 +1,27 @@
 package org.interledger.stream.calculators;
 
-import com.google.common.primitives.UnsignedLong;
 import org.interledger.stream.Denomination;
+
+import com.google.common.primitives.UnsignedLong;
+
 import java.util.Optional;
 
+/**
+ * Exchange rate calculator that ignores denominations and applies no rate calculations.
+ */
 public class NoOpExchangeRateCalculator implements ExchangeRateCalculator {
 
+
+  @Override
+  public UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive,
+                                            Denomination sendDenomination,
+                                            Denomination receiveDenomination) {
+    return amountToReceive;
+  }
+
   @Override
-  public UnsignedLong calculate(UnsignedLong sendAmount, Denomination sendDenomination,
-                                Optional<Denomination> expectedReceivedDenomination) {
+  public UnsignedLong calculateMinAmountToAccept(UnsignedLong sendAmount, Denomination sendDenomination,
+                                                 Optional<Denomination> expectedReceivedDenomination) {
     return UnsignedLong.ZERO;
   }
 
diff --git a/stream-parent/stream-core/src/main/java/org/interledger/stream/crypto/JavaxStreamEncryptionService.java b/stream-parent/stream-core/src/main/java/org/interledger/stream/crypto/JavaxStreamEncryptionService.java
index c085a43..aea9894 100644
--- a/stream-parent/stream-core/src/main/java/org/interledger/stream/crypto/JavaxStreamEncryptionService.java
+++ b/stream-parent/stream-core/src/main/java/org/interledger/stream/crypto/JavaxStreamEncryptionService.java
@@ -1,6 +1,7 @@
 package org.interledger.stream.crypto;
 
 import org.interledger.core.SharedSecret;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.hash.Hashing;
 
@@ -11,7 +12,6 @@ import java.security.InvalidKeyException;
 import java.security.NoSuchAlgorithmException;
 import java.util.Arrays;
 import java.util.Objects;
-
 import javax.crypto.BadPaddingException;
 import javax.crypto.Cipher;
 import javax.crypto.IllegalBlockSizeException;
diff --git a/stream-parent/stream-receiver/src/main/java/org/interledger/stream/receiver/StatelessStreamReceiver.java b/stream-parent/stream-receiver/src/main/java/org/interledger/stream/receiver/StatelessStreamReceiver.java
index ab22e41..0fe2340 100644
--- a/stream-parent/stream-receiver/src/main/java/org/interledger/stream/receiver/StatelessStreamReceiver.java
+++ b/stream-parent/stream-receiver/src/main/java/org/interledger/stream/receiver/StatelessStreamReceiver.java
@@ -1,5 +1,7 @@
 package org.interledger.stream.receiver;
 
+import static org.interledger.stream.FluentCompareTo.is;
+
 import org.interledger.core.InterledgerAddress;
 import org.interledger.core.InterledgerErrorCode;
 import org.interledger.core.InterledgerFulfillPacket;
@@ -138,7 +140,7 @@ public class StatelessStreamReceiver implements StreamReceiver {
     final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition());
 
     // Return Fulfill or Reject Packet
-    if (isFulfillable && preparePacket.getAmount().compareTo(streamPacket.prepareAmount()) >= 0) {
+    if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) {
       final StreamPacket returnableStreamPacketResponse = StreamPacket.builder()
           .sequence(streamPacket.sequence())
           .interledgerPacketType(InterledgerPacketType.FULFILL)
@@ -176,7 +178,7 @@ public class StatelessStreamReceiver implements StreamReceiver {
 
       if (isFulfillable) {
         logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket);
-      } else if (preparePacket.getAmount().compareTo(streamPacket.prepareAmount()) < 0) {
+      } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) {
         logger.debug(
             "Received only: {} when we should have received at least: {}",
             preparePacket.getAmount(), streamPacket.prepareAmount()


Test patch:
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTrackerTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTrackerTest.java
new file mode 100644
index 0000000..5c29489
--- /dev/null
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedReceiverAmountPaymentTrackerTest.java
@@ -0,0 +1,90 @@
+package org.interledger.stream.sender;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.interledger.stream.Denominations;
+import org.interledger.stream.PrepareAmounts;
+
+import com.google.common.primitives.UnsignedLong;
+import org.junit.Test;
+
+import java.util.Optional;
+
+public class FixedReceiverAmountPaymentTrackerTest {
+
+  @Test
+  public void checkAllInteractions() {
+    FixedReceiverAmountPaymentTracker tracker = new FixedReceiverAmountPaymentTracker(UnsignedLong.valueOf(10l),
+        new HalfsiesExchangeRateCalculator());
+
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    PrepareAmounts amounts = tracker.getSendPacketAmounts(UnsignedLong.ZERO, Denominations.XRP,
+        Optional.of(Denominations.XRP));
+
+    assertThat(amounts.getAmountToSend()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(amounts.getMinimumAmountToAccept()).isEqualTo(UnsignedLong.ZERO);
+
+    amounts = tracker.getSendPacketAmounts(UnsignedLong.valueOf(6), Denominations.XRP,
+        Optional.of(Denominations.XRP));
+
+    assertThat(amounts.getAmountToSend()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(amounts.getMinimumAmountToAccept()).isEqualTo(UnsignedLong.valueOf(3));
+
+    tracker.auth(amounts);
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(7));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.rollback(amounts, false);
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.auth(amounts);
+    tracker.commit(amounts, UnsignedLong.valueOf(3));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(3));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(7));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.auth(amounts);
+    tracker.commit(amounts, UnsignedLong.valueOf(4));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(7));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(3));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.auth(amounts);
+    tracker.commit(amounts, UnsignedLong.valueOf(3));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(18));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.moreToSend()).isFalse();
+
+    tracker.auth(amounts);
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(18));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(10));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.moreToSend()).isFalse();
+
+    amounts = tracker.getSendPacketAmounts(UnsignedLong.valueOf(6), Denominations.XRP,
+        Optional.of(Denominations.XRP));
+
+    assertThat(amounts.getAmountToSend()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(amounts.getMinimumAmountToAccept()).isEqualTo(UnsignedLong.ZERO);
+  }
+
+}
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedSenderAmountPaymentTrackerTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedSenderAmountPaymentTrackerTest.java
new file mode 100644
index 0000000..26455e2
--- /dev/null
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/FixedSenderAmountPaymentTrackerTest.java
@@ -0,0 +1,80 @@
+package org.interledger.stream.sender;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.interledger.stream.Denominations;
+import org.interledger.stream.PrepareAmounts;
+
+import com.google.common.primitives.UnsignedLong;
+import org.junit.Test;
+
+import java.util.Optional;
+
+public class FixedSenderAmountPaymentTrackerTest {
+
+  @Test
+  public void checkAllInteractions() {
+    FixedSenderAmountPaymentTracker tracker = new FixedSenderAmountPaymentTracker(UnsignedLong.valueOf(12l),
+        new HalfsiesExchangeRateCalculator());
+
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    PrepareAmounts amounts = tracker.getSendPacketAmounts(UnsignedLong.ZERO, Denominations.XRP,
+        Optional.of(Denominations.XRP));
+
+    assertThat(amounts.getAmountToSend()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(amounts.getMinimumAmountToAccept()).isEqualTo(UnsignedLong.ZERO);
+
+    amounts = tracker.getSendPacketAmounts(UnsignedLong.valueOf(6), Denominations.XRP,
+        Optional.of(Denominations.XRP));
+
+    assertThat(amounts.getAmountToSend()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(amounts.getMinimumAmountToAccept()).isEqualTo(UnsignedLong.valueOf(3));
+
+    tracker.auth(amounts);
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.rollback(amounts, false);
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.auth(amounts);
+    tracker.commit(amounts, UnsignedLong.valueOf(3));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(3));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.valueOf(6));
+    assertThat(tracker.moreToSend()).isTrue();
+
+    tracker.auth(amounts);
+    tracker.commit(amounts, UnsignedLong.valueOf(4));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(7));
+    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(tracker.moreToSend()).isFalse();
+
+    // FIXME add logic to prevent less than 0
+//    tracker.auth(amounts);
+//    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(12));
+//    assertThat(tracker.getDeliveredAmount()).isEqualTo(UnsignedLong.valueOf(7));
+//    assertThat(tracker.getOriginalAmount()).isEqualTo(UnsignedLong.valueOf(12));
+//    assertThat(tracker.getOriginalAmountLeft()).isEqualTo(UnsignedLong.ZERO);
+//    assertThat(tracker.moreToSend()).isFalse();
+
+    tracker.setSentAmount(UnsignedLong.valueOf(20));
+    assertThat(tracker.getAmountSent()).isEqualTo(UnsignedLong.valueOf(20));
+
+  }
+}
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/HalfsiesExchangeRateCalculator.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/HalfsiesExchangeRateCalculator.java
new file mode 100644
index 0000000..e3ec2ee
--- /dev/null
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/HalfsiesExchangeRateCalculator.java
@@ -0,0 +1,25 @@
+package org.interledger.stream.sender;
+
+import org.interledger.stream.Denomination;
+import org.interledger.stream.calculators.ExchangeRateCalculator;
+import org.interledger.stream.calculators.NoExchangeRateException;
+
+import com.google.common.primitives.UnsignedLong;
+
+import java.util.Optional;
+
+public class HalfsiesExchangeRateCalculator implements ExchangeRateCalculator {
+
+    @Override
+    public UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive, Denomination sendDenomination,
+                                              Denomination receiveDenomination) throws NoExchangeRateException {
+      return amountToReceive.times(UnsignedLong.valueOf(2));
+    }
+
+    @Override
+    public UnsignedLong calculateMinAmountToAccept(UnsignedLong sendAmount, Denomination sendDenomination,
+                                                   Optional<Denomination> expectedReceivedDenomination)
+        throws NoExchangeRateException {
+      return sendAmount.dividedBy(UnsignedLong.valueOf(2));
+    }
+  }
\ No newline at end of file
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 7867df2..8e95da6 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
@@ -21,7 +21,10 @@ 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.PaymentTracker;
+import org.interledger.stream.PrepareAmounts;
 import org.interledger.stream.SendMoneyRequest;
+import org.interledger.stream.SenderAmountMode;
 import org.interledger.stream.StreamConnection;
 import org.interledger.stream.StreamConnectionClosedException;
 import org.interledger.stream.StreamConnectionId;
@@ -54,7 +57,6 @@ 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}.
@@ -85,6 +87,8 @@ public class SendMoneyAggregatorTest {
   private UnsignedLong originalAmountToSend = UnsignedLong.valueOf(10L);
 
   private SendMoneyAggregator sendMoneyAggregator;
+  private PaymentTracker paymentTracker;
+  private PrepareAmounts defaultPrepareAmounts;
 
   @Before
   public void setUp() {
@@ -95,19 +99,22 @@ public class SendMoneyAggregatorTest {
     when(streamEncryptionServiceMock.decrypt(any(), any())).thenReturn(new byte[32]);
     when(linkMock.sendPacket(any())).thenReturn(mock(InterledgerRejectPacket.class));
     ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
+    this.paymentTracker = new FixedSenderAmountPaymentTracker(originalAmountToSend, new NoOpExchangeRateCalculator());
     SendMoneyRequest request = SendMoneyRequest.builder()
         .sharedSecret(sharedSecret)
         .sourceAddress(sourceAddress)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
         .destinationAddress(destinationAddress)
         .amount(originalAmountToSend)
         .timeout(Optional.of(Duration.ofSeconds(60)))
         .denomination(Denominations.XRP)
-        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .paymentTracker(paymentTracker)
         .build();
     this.sendMoneyAggregator = new SendMoneyAggregator(
         executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-        streamEncryptionServiceMock, request
-    );
+        streamEncryptionServiceMock, request);
+
+    defaultPrepareAmounts = PrepareAmounts.from(samplePreparePacket(), sampleStreamPacket());
   }
 
   @Test
@@ -206,11 +213,12 @@ public class SendMoneyAggregatorTest {
     SendMoneyRequest request = SendMoneyRequest.builder()
         .sharedSecret(sharedSecret)
         .sourceAddress(sourceAddress)
-        .destinationAddress(destinationAddress)
         .amount(originalAmountToSend)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+        .destinationAddress(destinationAddress)
         .timeout(Optional.of(Duration.ofSeconds(60)))
         .denomination(Denominations.XRP)
-        .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+        .paymentTracker(new FixedSenderAmountPaymentTracker(originalAmountToSend, new NoOpExchangeRateCalculator()))
         .build();
     ExecutorService executor = mock(ExecutorService.class);
     this.sendMoneyAggregator = new SendMoneyAggregator(
@@ -229,7 +237,8 @@ public class SendMoneyAggregatorTest {
         .build();
 
     expectedException.expect(RejectedExecutionException.class);
-    sendMoneyAggregator.schedule(new AtomicBoolean(false), prepare, sampleStreamPacket(), UnsignedLong.ONE);
+    sendMoneyAggregator.schedule(new AtomicBoolean(false), prepare, sampleStreamPacket(),
+        PrepareAmounts.from(prepare, sampleStreamPacket()));
     verify(congestionControllerMock, times(1)).reject(UnsignedLong.ONE, expectedReject);
   }
 
@@ -340,59 +349,58 @@ public class SendMoneyAggregatorTest {
   public void handleRejectHatesNullPrepare() {
     expectedException.expect(NullPointerException.class);
     sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
-        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
-        new AtomicReference<>(), congestionControllerMock);
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), defaultPrepareAmounts, new AtomicInteger(),
+        congestionControllerMock);
   }
 
   @Test
   public void handleRejectHatesNullStreamPacket() {
     expectedException.expect(NullPointerException.class);
     sendMoneyAggregator.handleReject(samplePreparePacket(), null,
-        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
-        new AtomicReference<>(), congestionControllerMock);
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), defaultPrepareAmounts, new AtomicInteger(),
+        congestionControllerMock);
   }
 
   @Test
   public void handleRejectHatesNullReject() {
     expectedException.expect(NullPointerException.class);
-    sendMoneyAggregator.handleReject(null, sampleStreamPacket(), null, new AtomicInteger(),
-        new AtomicReference<UnsignedLong>(), congestionControllerMock);
+    sendMoneyAggregator.handleReject(null, sampleStreamPacket(), null, defaultPrepareAmounts,
+        new AtomicInteger(), congestionControllerMock);
   }
 
   @Test
   public void handleRejectHatesNullNumReject() {
     expectedException.expect(NullPointerException.class);
     sendMoneyAggregator.handleReject(null, sampleStreamPacket(), sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR),
-        null, new AtomicReference<UnsignedLong>(), congestionControllerMock);
+        null, null, congestionControllerMock);
   }
 
   @Test
-  public void handleRejectHatesNullAmountLeftToSend() {
+  public void handleRejectHatesNullCongestionController() {
     expectedException.expect(NullPointerException.class);
     sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
-        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
-        null, congestionControllerMock);
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), defaultPrepareAmounts, new AtomicInteger(),
+        null);
   }
 
   @Test
-  public void handleRejectHatesNullCongestionController() {
+  public void handleRejectHatesNullPrepareAmountsr() {
     expectedException.expect(NullPointerException.class);
     sendMoneyAggregator.handleReject(null, sampleStreamPacket(),
-        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), new AtomicInteger(),
-        new AtomicReference<UnsignedLong>(), null);
+        sampleRejectPacket(InterledgerErrorCode.T00_INTERNAL_ERROR), null, new AtomicInteger(),
+        congestionControllerMock);
   }
 
   @Test
   public void handleReject() {
+    UnsignedLong originalAmountToSend = paymentTracker.getOriginalAmountLeft();
     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);
+        reject, defaultPrepareAmounts, numReject, congestionControllerMock);
     assertThat(numReject.get()).isEqualTo(1);
-    assertThat(amountLeftToSend.get().intValue()).isEqualTo(2);
+    assertThat(paymentTracker.getOriginalAmountLeft()).isEqualTo(originalAmountToSend.plus(prepare.getAmount()));
     verify(congestionControllerMock, times(1)).reject(UnsignedLong.ONE, reject);
   }
 
@@ -405,10 +413,11 @@ public class SendMoneyAggregatorTest {
   ) {
     when(congestionControllerMock.hasInFlight()).thenReturn(moneyInFlight);
     when(streamConnectionMock.isClosed()).thenReturn(streamConnectionClosed);
+    FixedSenderAmountPaymentTracker fixedSender = (FixedSenderAmountPaymentTracker) this.paymentTracker;
     if (moreToSend) {
-      sendMoneyAggregator.setAmountSentForTesting(UnsignedLong.ZERO);
+      fixedSender.setSentAmount(UnsignedLong.ZERO);
     } else {
-      sendMoneyAggregator.setAmountSentForTesting(UnsignedLong.valueOf(10L));
+      fixedSender.setSentAmount(UnsignedLong.valueOf(10L));
     }
   }
 
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 8e6c7b5..bc28375 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,11 +2,9 @@ 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;
-import org.interledger.core.SharedSecret;
 import org.interledger.link.Link;
 import org.interledger.link.LinkId;
 import org.interledger.link.http.IlpOverHttpLink;
@@ -26,8 +24,10 @@ import org.interledger.stream.Denomination;
 import org.interledger.stream.Denominations;
 import org.interledger.stream.SendMoneyRequest;
 import org.interledger.stream.SendMoneyResult;
+import org.interledger.stream.SenderAmountMode;
 import org.interledger.stream.calculators.ExchangeRateCalculator;
 import org.interledger.stream.calculators.NoExchangeRateException;
+import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
 import org.interledger.stream.crypto.JavaxStreamEncryptionService;
 
 import com.fasterxml.jackson.core.JsonGenerator;
@@ -66,7 +66,6 @@ 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;
@@ -164,8 +163,8 @@ public class SimpleStreamSenderIT {
   }
 
   /**
-   * One call to {@link SimpleStreamSender#sendMoney(SharedSecret, InterledgerAddress, InterledgerAddress,
-   * UnsignedLong, Denomination)} that involves a single packet for the entire payment.
+   * One call to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)} that involves a single packet for the entire
+   * payment.
    */
   @Test
   public void sendMoneySinglePacket() {
@@ -177,12 +176,50 @@ public class SimpleStreamSenderIT {
 
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(1000000);
 
+    SendMoneyRequest request = SendMoneyRequest.builder()
+        .sourceAddress(SENDER_ADDRESS)
+        .amount(paymentAmount)
+        .denomination(Denominations.XRP)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+        .destinationAddress(connectionDetails.destinationAddress())
+        .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+        .sharedSecret(connectionDetails.sharedSecret())
+        .build();
+
+    final SendMoneyResult sendMoneyResult = streamSender.sendMoney(request).join();
+
+    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);
+  }
+
+  /**
+   * One call to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)} that involves a single packet for the entire
+   * payment.
+   */
+  @Test
+  public void sendSmallPayment() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(100);
+
+    StreamSender streamSender = new SimpleStreamSender(
+        new JavaxStreamEncryptionService(), link
+    );
+
+    final StreamConnectionDetails connectionDetails = getStreamConnectionDetails("sendSmallPayment");
+
     final SendMoneyResult sendMoneyResult = streamSender.sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount,
-        Denominations.XRP
+        SendMoneyRequest.builder()
+            .sourceAddress(SENDER_ADDRESS)
+            .amount(paymentAmount)
+            .denomination(Denominations.XRP)
+            .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+            .destinationAddress(connectionDetails.destinationAddress())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+            .sharedSecret(connectionDetails.sharedSecret())
+            .build()
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -214,11 +251,15 @@ public class SimpleStreamSenderIT {
 
     for (int i = 0; i < numExecutions; i++) {
       final CompletableFuture<SendMoneyResult> job = streamSender.sendMoney(
-          SharedSecret.of(connectionDetails.sharedSecret().key()),
-          SENDER_ADDRESS,
-          connectionDetails.destinationAddress(),
-          paymentAmount,
-          Denominations.XRP
+          SendMoneyRequest.builder()
+              .sourceAddress(SENDER_ADDRESS)
+              .amount(paymentAmount)
+              .denomination(Denominations.XRP)
+              .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+              .destinationAddress(connectionDetails.destinationAddress())
+              .sharedSecret(connectionDetails.sharedSecret())
+              .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+              .build()
       );
       results.add(job);
     }
@@ -228,7 +269,7 @@ public class SimpleStreamSenderIT {
     CompletableFuture.allOf(completableFutures)
         .handle(($, error) -> {
           if (error != null) {
-            logger.error("ERROR: {}", error);
+            logger.error(error.getMessage(), error);
           }
           // To placate completable future.
           return null;
@@ -259,8 +300,7 @@ public class SimpleStreamSenderIT {
   }
 
   /**
-   * One call to {@link SimpleStreamSender#sendMoney(SharedSecret, InterledgerAddress, InterledgerAddress,
-   * UnsignedLong)} that involves multiple packets in parallel.
+   * One call to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)}} that involves multiple packets in parallel.
    */
   @Test
   public void sendMoneyMultiPacket() {
@@ -273,11 +313,15 @@ public class SimpleStreamSenderIT {
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(1000001);
 
     final SendMoneyResult sendMoneyResult = streamSender.sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount,
-        Denominations.XRP
+        SendMoneyRequest.builder()
+            .sourceAddress(SENDER_ADDRESS)
+            .amount(paymentAmount)
+            .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+            .denomination(Denominations.XRP)
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+            .build()
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -289,9 +333,8 @@ public class SimpleStreamSenderIT {
   }
 
   /**
-   * Multiple calls to {@link SimpleStreamSender#sendMoney(SharedSecret, InterledgerAddress, InterledgerAddress,
-   * UnsignedLong)} that involves multiple packets in parallel, but using different accounts for each Stream, and thus a
-   * different Connection.
+   * Multiple calls to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)}} that involves multiple packets in
+   * parallel, but using different accounts for each Stream, and thus a different Connection.
    */
   @Test
   public void sendMoneyMultiThreadedToSeparateAccounts() throws InterruptedException {
@@ -358,12 +401,18 @@ public class SimpleStreamSenderIT {
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(account);
 
     final SendMoneyResult sendMoneyResult = streamSender
-        .sendMoney(SharedSecret.of(connectionDetails.sharedSecret().key()),
-            SENDER_ADDRESS,
-            connectionDetails.destinationAddress(),
-            paymentAmount,
-            Denominations.XRP,
-            Duration.ofMillis(100)).join();
+        .sendMoney(
+            SendMoneyRequest.builder()
+                .sourceAddress(SENDER_ADDRESS)
+                .amount(paymentAmount)
+                .denomination(Denominations.XRP)
+                .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+                .destinationAddress(connectionDetails.destinationAddress())
+                .sharedSecret(connectionDetails.sharedSecret())
+                .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+                .timeout(Duration.ofMillis(100))
+                .build()
+        ).join();
 
     assertThat(sendMoneyResult.successfulPayment()).isFalse();
 
@@ -390,18 +439,16 @@ public class SimpleStreamSenderIT {
 
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(account);
 
-    ExchangeRateCalculator noExchangeRateExceptionCalculator =
-        (sendAmount, sendDenomination, expectedReceivedDenomination) -> {
-      throw new NoExchangeRateException("Never stop never calculating");
-    };
+    ExchangeRateCalculator noExchangeRateExceptionCalculator = new CrankyExchangeRateCalculator();
 
     SendMoneyRequest request = SendMoneyRequest.builder()
+        .sourceAddress(SENDER_ADDRESS)
         .amount(paymentAmount)
         .denomination(Denominations.XRP)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
         .destinationAddress(connectionDetails.destinationAddress())
-        .exchangeRateCalculator(noExchangeRateExceptionCalculator)
+        .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, noExchangeRateExceptionCalculator))
         .sharedSecret(connectionDetails.sharedSecret())
-        .sourceAddress(SENDER_ADDRESS)
         .timeout(Duration.ofMillis(100))
         .build();
 
@@ -427,17 +474,14 @@ public class SimpleStreamSenderIT {
 
     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()
+        .sourceAddress(SENDER_ADDRESS)
         .amount(paymentAmount)
         .denomination(Denominations.XRP)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
         .destinationAddress(connectionDetails.destinationAddress())
-        .exchangeRateCalculator(reasonableCalculator)
+        .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
         .sharedSecret(connectionDetails.sharedSecret())
-        .sourceAddress(SENDER_ADDRESS)
         .timeout(Duration.ofMillis(1000))
         .build();
 
@@ -461,17 +505,14 @@ public class SimpleStreamSenderIT {
 
     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()
+        .sourceAddress(SENDER_ADDRESS)
         .amount(paymentAmount)
         .denomination(Denominations.XRP)
+        .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
         .destinationAddress(connectionDetails.destinationAddress())
-        .exchangeRateCalculator(unreasonableCalculator)
+        .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new GreedyExchangeRateCalculator()))
         .sharedSecret(connectionDetails.sharedSecret())
-        .sourceAddress(SENDER_ADDRESS)
         .timeout(Duration.ofMillis(1000))
         .build();
 
@@ -522,15 +563,17 @@ public class SimpleStreamSenderIT {
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(1000000);
 
     streamSender.sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount,
-        Denominations.XRP,
-        Duration.ofMillis(100L)
-    ).whenComplete(($, error) -> {
-      assertThat(error).isNotNull();
-    });
+        SendMoneyRequest.builder()
+            .sourceAddress(SENDER_ADDRESS)
+            .destinationAddress(connectionDetails.destinationAddress())
+            .amount(paymentAmount)
+            .denomination(Denominations.XRP)
+            .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+            .timeout(Duration.ofMillis(100L))
+            .build()
+    ).whenComplete(($, error) -> assertThat(error).isNotNull());
   }
 
   private StreamConnectionDetails getStreamConnectionDetails(int id) {
@@ -563,11 +606,15 @@ public class SimpleStreamSenderIT {
     );
 
     final SendMoneyResult sendMoneyResult = streamSender.sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount,
-        Denominations.XRP
+        SendMoneyRequest.builder()
+            .sourceAddress(SENDER_ADDRESS)
+            .destinationAddress(connectionDetails.destinationAddress())
+            .amount(paymentAmount)
+            .denomination(Denominations.XRP)
+            .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+            .build()
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -607,4 +654,35 @@ public class SimpleStreamSenderIT {
         .writeTimeout(30, TimeUnit.SECONDS);
     return builder.connectionPool(connectionPool).build();
   }
+
+  static class CrankyExchangeRateCalculator implements ExchangeRateCalculator {
+
+    @Override
+    public UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive, Denomination sendDenomination,
+        Denomination receiveDenomination) {
+      throw new NoExchangeRateException("no exchanges allowed");
+    }
+
+    @Override
+    public UnsignedLong calculateMinAmountToAccept(UnsignedLong sendAmount, Denomination sendDenomination,
+        Optional<Denomination> expectedReceivedDenomination) throws NoExchangeRateException {
+      throw new NoExchangeRateException("no exchanges allowed");
+    }
+  }
+
+  static class GreedyExchangeRateCalculator implements ExchangeRateCalculator {
+
+    @Override
+    public UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive, Denomination sendDenomination,
+        Denomination receiveDenomination) {
+      return amountToReceive.plus(UnsignedLong.ONE);
+    }
+
+    @Override
+    public UnsignedLong calculateMinAmountToAccept(UnsignedLong sendAmount, Denomination sendDenomination,
+        Optional<Denomination> expectedReceivedDenomination) throws NoExchangeRateException {
+      return sendAmount.plus(UnsignedLong.ONE);
+    }
+  }
+
 }
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
index 2a9b42d..30d1d56 100644
--- 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
@@ -1,9 +1,9 @@
 package org.interledger.stream;
 
-import org.junit.Test;
-
 import static org.assertj.core.api.Assertions.assertThat;
 
+import org.junit.Test;
+
 public class DenominationsTest {
 
   @Test
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/FluentCompareToTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/FluentCompareToTest.java
new file mode 100644
index 0000000..57225be
--- /dev/null
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/FluentCompareToTest.java
@@ -0,0 +1,65 @@
+package org.interledger.stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.interledger.stream.FluentCompareTo.is;
+
+import com.google.common.primitives.UnsignedLong;
+import org.junit.Test;
+
+public class FluentCompareToTest {
+
+  @Test
+  public void isEqualTo() {
+    assertThat(is(UnsignedLong.ZERO).equalTo(UnsignedLong.ZERO)).isTrue();
+    assertThat(is(UnsignedLong.ZERO).equalTo(UnsignedLong.ONE)).isFalse();
+    assertThat(is("foo").equalTo("foo")).isTrue();
+    assertThat(is("foo").equalTo("bar")).isFalse();
+  }
+
+  @Test
+  public void isLessThan() {
+    assertThat(is(UnsignedLong.ZERO).lessThan(UnsignedLong.ZERO)).isFalse();
+    assertThat(is(UnsignedLong.ZERO).lessThan(UnsignedLong.ONE)).isTrue();
+    assertThat(is(UnsignedLong.ONE).lessThan(UnsignedLong.ZERO)).isFalse();
+  }
+
+  @Test
+  public void isLessThanOrEqualTo() {
+    assertThat(is(UnsignedLong.ZERO).lessThanOrEqualTo(UnsignedLong.ZERO)).isTrue();
+    assertThat(is(UnsignedLong.ZERO).lessThanOrEqualTo(UnsignedLong.ONE)).isTrue();
+    assertThat(is(UnsignedLong.ONE).lessThanOrEqualTo(UnsignedLong.ZERO)).isFalse();
+  }
+
+  @Test
+  public void isGreaterThan() {
+    assertThat(is(UnsignedLong.ZERO).greaterThan(UnsignedLong.ZERO)).isFalse();
+    assertThat(is(UnsignedLong.ZERO).greaterThan(UnsignedLong.ONE)).isFalse();
+    assertThat(is(UnsignedLong.ONE).greaterThan(UnsignedLong.ZERO)).isTrue();
+  }
+
+  @Test
+  public void isGreaterThanOrEqualTo() {
+    assertThat(is(UnsignedLong.ZERO).greaterThanEqualTo(UnsignedLong.ZERO)).isTrue();
+    assertThat(is(UnsignedLong.ZERO).greaterThanEqualTo(UnsignedLong.ONE)).isFalse();
+    assertThat(is(UnsignedLong.ONE).greaterThanEqualTo(UnsignedLong.ZERO)).isTrue();
+  }
+
+  @Test
+  public void isBetween() {
+    assertThat(is(UnsignedLong.ZERO).between(UnsignedLong.ZERO, UnsignedLong.ONE)).isTrue();
+    assertThat(is(UnsignedLong.ONE).between(UnsignedLong.ZERO, UnsignedLong.MAX_VALUE)).isTrue();
+    assertThat(is(UnsignedLong.ONE).between(UnsignedLong.ZERO, UnsignedLong.ONE)).isFalse();
+    assertThat(is(UnsignedLong.ZERO).between(UnsignedLong.ONE, UnsignedLong.MAX_VALUE)).isFalse();
+    assertThat(is(UnsignedLong.MAX_VALUE).between(UnsignedLong.ZERO, UnsignedLong.ONE)).isFalse();
+  }
+
+  @Test
+  public void isBetweenExclusive() {
+    assertThat(is(UnsignedLong.ZERO).betweenExclusive(UnsignedLong.ZERO, UnsignedLong.ONE)).isFalse();
+    assertThat(is(UnsignedLong.ONE).betweenExclusive(UnsignedLong.ZERO, UnsignedLong.MAX_VALUE)).isTrue();
+    assertThat(is(UnsignedLong.ONE).betweenExclusive(UnsignedLong.ZERO, UnsignedLong.ONE)).isFalse();
+    assertThat(is(UnsignedLong.ZERO).betweenExclusive(UnsignedLong.ONE, UnsignedLong.MAX_VALUE)).isFalse();
+    assertThat(is(UnsignedLong.MAX_VALUE).betweenExclusive(UnsignedLong.ZERO, UnsignedLong.ONE)).isFalse();
+  }
+
+}
\ No newline at end of file
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/PrepareAmountsTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/PrepareAmountsTest.java
new file mode 100644
index 0000000..c0b21a5
--- /dev/null
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/PrepareAmountsTest.java
@@ -0,0 +1,31 @@
+package org.interledger.stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.interledger.core.InterledgerPreparePacket;
+
+import com.google.common.primitives.UnsignedLong;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link PrepareAmounts}.
+ */
+public class PrepareAmountsTest {
+
+  @Test
+  public void testFrom() {
+    final UnsignedLong amountToSend = UnsignedLong.MAX_VALUE;
+    InterledgerPreparePacket preparePacketMock = mock(InterledgerPreparePacket.class);
+    when(preparePacketMock.getAmount()).thenReturn(amountToSend);
+
+    final UnsignedLong minimumAmountToAccept = UnsignedLong.ONE;
+    StreamPacket streamPacketMock = mock(StreamPacket.class);
+    when(streamPacketMock.prepareAmount()).thenReturn(minimumAmountToAccept);
+
+    PrepareAmounts actual = PrepareAmounts.from(preparePacketMock, streamPacketMock);
+    assertThat(actual.getAmountToSend()).isEqualTo(amountToSend);
+    assertThat(actual.getMinimumAmountToAccept()).isEqualTo(minimumAmountToAccept);
+  }
+}
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 c3468a2..4336eca 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
@@ -22,6 +22,7 @@ public class SendMoneyResultTest {
         .numFulfilledPackets(10)
         .numRejectPackets(5)
         .sendMoneyDuration(Duration.ZERO)
+        .successfulPayment(true)
         .build();
     assertThat(result.totalPackets()).isEqualTo(15);
   }
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
index c92ed69..2fbb32a 100644
--- 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
@@ -1,14 +1,15 @@
 package org.interledger.stream.calculators;
 
-import com.google.common.primitives.UnsignedLong;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+
 import org.interledger.stream.Denominations;
+
+import com.google.common.primitives.UnsignedLong;
 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
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
index 269527e..81a82f1 100644
--- 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
@@ -1,18 +1,20 @@
 package org.interledger.stream.calculators;
 
-import com.google.common.primitives.UnsignedLong;
+import static org.assertj.core.api.Assertions.assertThat;
+
 import org.interledger.stream.Denominations;
+
+import com.google.common.primitives.UnsignedLong;
 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);
+    assertThat(calc.calculateMinAmountToAccept(UnsignedLong.ONE, Denominations.USD, Optional.of(Denominations.EUR)))
+        .isEqualTo(UnsignedLong.ZERO);
   }
 }
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/JavaxStreamEncryptionServiceTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/JavaxStreamEncryptionServiceTest.java
index 0a75577..9c7a4d4 100644
--- a/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/JavaxStreamEncryptionServiceTest.java
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/JavaxStreamEncryptionServiceTest.java
@@ -3,6 +3,7 @@ package org.interledger.stream.crypto;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import org.interledger.core.SharedSecret;
+
 import org.junit.Before;
 import org.junit.Test;
 
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/SharedSecretTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/SharedSecretTest.java
index 31091d8..ad4e389 100644
--- a/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/SharedSecretTest.java
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/crypto/SharedSecretTest.java
@@ -3,6 +3,7 @@ package org.interledger.stream.crypto;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import org.interledger.core.SharedSecret;
+
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
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 9a063aa..2024f9a 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
@@ -1,22 +1,25 @@
 package org.interledger.stream.receiver;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.interledger.stream.SenderAmountMode.RECEIVER_AMOUNT;
+import static org.interledger.stream.SenderAmountMode.SENDER_AMOUNT;
 
 import org.interledger.codecs.stream.StreamCodecContextFactory;
 import org.interledger.core.Immutable;
 import org.interledger.core.InterledgerAddress;
-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.ExchangeRateCalculator;
 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;
 import org.interledger.stream.receiver.testutils.SimulatedPathConditions;
+import org.interledger.stream.sender.FixedReceiverAmountPaymentTracker;
+import org.interledger.stream.sender.FixedSenderAmountPaymentTracker;
 import org.interledger.stream.sender.SimpleStreamSender;
 import org.interledger.stream.sender.StreamSender;
 
@@ -30,9 +33,12 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.time.Duration;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -63,6 +69,41 @@ public class SenderReceiverTest {
 
   private SimulatedIlpv4Network simulatedIlpNetwork;
 
+  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, link
+    );
+
+    StatelessStreamReceiver streamReceiver = new StatelessStreamReceiver(
+        () -> serverSecret,
+        new SpspStreamConnectionGenerator(),
+        streamEncryptionService,
+        StreamCodecContextFactory.oer()
+    );
+
+    return StreamNode.builder()
+        .serverSecret(serverSecret)
+        .senderAddress(senderAddress)
+        .receiverAddress(receiverAddress)
+        .streamSender(streamSender)
+        .streamReceiver(streamReceiver)
+        .denomination(Denomination.builder().assetCode("XRP").assetScale((short) 6).build())
+        .link(link)
+        .build();
+  }
+
+  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());
+  }
+
   @Before
   public void setup() {
     this.initIlpNetworkForStream();
@@ -72,7 +113,6 @@ public class SenderReceiverTest {
   public void testSendFromLeftToRight() {
     final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
 
-    final StreamConnectionDetails connectionDetails = leftStreamNode.getNewStreamConnectionDetails();
     final SendMoneyResult sendMoneyResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -83,17 +123,143 @@ public class SenderReceiverTest {
     logger.info("Payment Sent: {}", sendMoneyResult);
   }
 
+  @Test
+  public void testSendFromLeftToRightWithFixedReceiverAmountExactPathRate() {
+    final UnsignedLong amountToDeliver = UnsignedLong.valueOf(10000);
+
+    BigDecimal expectedPathRate = new BigDecimal("1.5");
+    BigDecimal actualPathRate = new BigDecimal("1.5");
+    this.initIlpNetworkForStream(new SimulatedIlpv4Network(
+        SimulatedPathConditions.builder().currentExchangeRateSupplier(() -> actualPathRate).build(),
+        SimulatedPathConditions.builder().build()
+    ));
+
+    SimpleStreamSender sender = new SimpleStreamSender(leftStreamNode.link());
+    final StreamConnectionDetails connectionDetails = rightStreamNode.getNewStreamConnectionDetails();
+
+    SendMoneyResult sendMoneyResult = sender.sendMoney(
+        SendMoneyRequest.builder()
+            .sourceAddress(leftStreamNode.senderAddress())
+            .amount(amountToDeliver)
+            .senderAmountMode(RECEIVER_AMOUNT)
+            .denomination(leftStreamNode.denomination())
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(
+                new FixedReceiverAmountPaymentTracker(amountToDeliver,
+                    new FixedRateExchangeCalculator(expectedPathRate))
+            )
+            .timeout(Duration.ofMillis(1000L))
+            .build())
+        .join();
+
+    assertThat(sendMoneyResult.amountDelivered()).isEqualTo(amountToDeliver);
+    assertThat(sendMoneyResult.originalAmount()).isEqualTo(amountToDeliver);
+    assertThat(sendMoneyResult.numFulfilledPackets()).isGreaterThan(1);
+    assertThat(sendMoneyResult.numRejectPackets()).isEqualTo(0);
+    assertThat(sendMoneyResult.successfulPayment()).isTrue();
+
+    logger.info("Payment Sent: {}", sendMoneyResult);
+  }
+
+  @Test
+  public void testSendFromLeftToRightWithFixedReceiverAmountOverestimatedPathRate() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(10000);
+
+    // simulating get a better exchange rate than we expected in which case payments should get fulfilled because
+    // the receiver gets more than they were supposed to
+    BigDecimal expectedPathExchange = new BigDecimal("1.4");
+    BigDecimal actualPathExchange = new BigDecimal("1.5");
+    this.initIlpNetworkForStream(new SimulatedIlpv4Network(
+        SimulatedPathConditions.builder().currentExchangeRateSupplier(() -> actualPathExchange).build(),
+        SimulatedPathConditions.builder().build()
+    ));
+
+    SimpleStreamSender sender = new SimpleStreamSender(leftStreamNode.link());
+    final StreamConnectionDetails connectionDetails = rightStreamNode.getNewStreamConnectionDetails();
+
+    SendMoneyResult sendMoneyResult = sender.sendMoney(
+        SendMoneyRequest.builder()
+            .sharedSecret(connectionDetails.sharedSecret())
+            .amount(paymentAmount)
+            .senderAmountMode(RECEIVER_AMOUNT)
+            .denomination(leftStreamNode.denomination())
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sourceAddress(leftStreamNode.senderAddress())
+            .paymentTracker(
+                new FixedReceiverAmountPaymentTracker(paymentAmount,
+                    new FixedRateExchangeCalculator(expectedPathExchange))
+            )
+            .timeout(Duration.ofMillis(1000L))
+            .build()
+    ).join();
+
+    assertThat(sendMoneyResult.amountDelivered()).isGreaterThanOrEqualTo(paymentAmount);
+    assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
+    assertThat(sendMoneyResult.amountSent()).isLessThan(paymentAmount);
+    assertThat(sendMoneyResult.numFulfilledPackets()).isGreaterThan(1);
+    assertThat(sendMoneyResult.numRejectPackets()).isEqualTo(0);
+    assertThat(sendMoneyResult.successfulPayment()).isTrue();
+
+    logger.info("Payment Sent: {}", sendMoneyResult);
+  }
+
+  @Test
+  public void testSendFromLeftToRightWithFixedReceiverAmountUnderestimatedPathRate() {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(10000);
+
+    // simulating get a worse exchange rate than we expected in which case payments should get rejected because
+    // the receiver gets less than they were supposed to
+    BigDecimal expectedPathExchange = new BigDecimal("1.5");
+    BigDecimal actualPathExchange = new BigDecimal("1.4"); // lower rate than expected
+    this.initIlpNetworkForStream(new SimulatedIlpv4Network(
+        SimulatedPathConditions.builder().currentExchangeRateSupplier(() -> actualPathExchange).build(),
+        SimulatedPathConditions.builder().build()
+    ));
+
+    SimpleStreamSender sender = new SimpleStreamSender(leftStreamNode.link());
+    final StreamConnectionDetails connectionDetails = rightStreamNode.getNewStreamConnectionDetails();
+
+    SendMoneyResult sendMoneyResult = sender.sendMoney(
+        SendMoneyRequest.builder()
+            .sourceAddress(leftStreamNode.senderAddress())
+            .amount(paymentAmount)
+            .senderAmountMode(RECEIVER_AMOUNT)
+            .denomination(leftStreamNode.denomination())
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedReceiverAmountPaymentTracker(paymentAmount,
+                new FixedRateExchangeCalculator(expectedPathExchange)))
+            .timeout(Duration.ofMillis(1000))
+            .build())
+        .join();
+
+    assertThat(sendMoneyResult.amountDelivered()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
+    assertThat(sendMoneyResult.amountSent()).isEqualTo(UnsignedLong.ZERO);
+    assertThat(sendMoneyResult.numFulfilledPackets()).isEqualTo(0);
+    assertThat(sendMoneyResult.numRejectPackets()).isGreaterThan(0);
+    assertThat(sendMoneyResult.successfulPayment()).isFalse();
+
+    logger.info("Payment Sent: {}", sendMoneyResult);
+  }
+
   @Test
   public void testSendFromRightToLeft() {
     final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
 
-    final StreamConnectionDetails connectionDetails = rightStreamNode.getNewStreamConnectionDetails();
+    final StreamConnectionDetails connectionDetails = leftStreamNode.getNewStreamConnectionDetails();
     final SendMoneyResult sendMoneyResult = rightStreamNode.streamSender().sendMoney(
-        SharedSecret.of(connectionDetails.sharedSecret().key()),
-        RIGHT_SENDER_ADDRESS,
-        connectionDetails.destinationAddress(),
-        paymentAmount,
-        Denominations.XRP
+        SendMoneyRequest.builder()
+            .sourceAddress(rightStreamNode.senderAddress())
+            .amount(paymentAmount)
+            .senderAmountMode(SENDER_AMOUNT)
+            .denomination(rightStreamNode.denomination())
+            .destinationAddress(connectionDetails.destinationAddress())
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+            .timeout(Duration.ofMillis(10000))
+            .build()
     ).join();
 
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
@@ -173,6 +339,7 @@ public class SenderReceiverTest {
     this.initIlpNetworkForStream(simulatedIlpNetwork);
 
     final SendMoneyResult sendMoneyResult = sendMoney(leftStreamNode, rightStreamNode, paymentAmount);
+
     assertThat(sendMoneyResult.amountDelivered()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.originalAmount()).isEqualTo(paymentAmount);
     assertThat(sendMoneyResult.numFulfilledPackets()).isEqualTo(100);
@@ -241,29 +408,32 @@ public class SenderReceiverTest {
     });
   }
 
+  /////////////////
+  // Helper Methods
+  /////////////////
+
   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) {
+  private SendMoneyResult sendMoney(
+      StreamSender sender, StreamNode fromNode, StreamNode toNode, UnsignedLong paymentAmount
+  ) {
     final StreamConnectionDetails connectionDetails = toNode.getNewStreamConnectionDetails();
     return sender.sendMoney(
         SendMoneyRequest.builder()
-            .sharedSecret(connectionDetails.sharedSecret())
+            .sourceAddress(fromNode.senderAddress())
             .amount(paymentAmount)
+            .senderAmountMode(SENDER_AMOUNT)
             .denomination(fromNode.denomination())
             .destinationAddress(connectionDetails.destinationAddress())
-            .sourceAddress(fromNode.senderAddress())
-            .exchangeRateCalculator(new NoOpExchangeRateCalculator())
+            .sharedSecret(connectionDetails.sharedSecret())
+            .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
             .timeout(Duration.ofMillis(10000))
             .build())
         .join();
   }
 
-  /////////////////
-  // Helper Methods
-  /////////////////
-
   /**
    * Helper method to test lossy ILPv4 network percentages.
    */
@@ -310,39 +480,11 @@ public class SenderReceiverTest {
   }
 
   private StreamNode initLeftNode() {
-    return initNode(simulatedIlpNetwork.getLeftToRightLink(), LEFT_SENDER_ADDRESS, RIGHT_RECEIVER_ADDRESS);
+    return initNode(simulatedIlpNetwork.getLeftToRightLink(), LEFT_SENDER_ADDRESS, LEFT_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, link
-    );
-
-    StatelessStreamReceiver streamReceiver = new StatelessStreamReceiver(
-        () -> serverSecret,
-        new SpspStreamConnectionGenerator(),
-        streamEncryptionService,
-        StreamCodecContextFactory.oer()
-    );
-
-    return StreamNode.builder()
-        .serverSecret(serverSecret)
-        .senderAddress(senderAddress)
-        .receiverAddress(receiverAddress)
-        .streamSender(streamSender)
-        .streamReceiver(streamReceiver)
-        .denomination(Denomination.builder().assetCode("XRP").assetScale((short) 6).build())
-        .link(link)
-        .build();
+    return initNode(simulatedIlpNetwork.getRightToLeftLink(), RIGHT_SENDER_ADDRESS, RIGHT_RECEIVER_ADDRESS);
   }
 
   private <T> List<CompletableFuture<T>> runInParallel(int parallelism, int runCount, Callable<T> task) {
@@ -364,11 +506,6 @@ public class SenderReceiverTest {
     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.
@@ -403,23 +540,65 @@ public class SenderReceiverTest {
 
     /**
      * The {@link Link} that this node uses to communicate to its immediate peer with.
+     *
+     * @return the {@link Link}.
      */
     Link link();
 
     /**
      * The {@link StreamSender} that this node uses to send STREAM payments.
+     *
+     * @return The {@link StreamSender}.
      */
     StreamSender streamSender();
 
     /**
      * The {@link StreamReceiver} that this node uses to receive STREAM payments on.
+     *
+     * @return The {@link StreamReceiver}.
      */
     StreamReceiver streamReceiver();
 
+    /**
+     * Constructs new {@link StreamConnectionDetails} for this node.
+     *
+     * @return An instance of {@link StreamConnectionDetails}.
+     */
     @Derived
     default StreamConnectionDetails getNewStreamConnectionDetails() {
       return new SpspStreamConnectionGenerator().generateConnectionDetails(this::serverSecret, receiverAddress());
     }
 
   }
+
+  /**
+   * An {@link ExchangeRateCalculator} that specifies a fixed exchange rate forever.
+   */
+  static class FixedRateExchangeCalculator implements ExchangeRateCalculator {
+
+    // exchange rate from sender unit's to receiver unit's
+    // if rate = 0.9 then 1 sender unit = 0.9 receiver units
+    private final BigDecimal senderUnitsPerReceiverUnits;
+
+    FixedRateExchangeCalculator(BigDecimal senderUnitsPerReceiverUnits) {
+      this.senderUnitsPerReceiverUnits = senderUnitsPerReceiverUnits;
+    }
+
+    @Override
+    public UnsignedLong calculateAmountToSend(UnsignedLong amountToReceive,
+        Denomination sendDenomination,
+        Denomination receiveDenomination) {
+      return UnsignedLong.valueOf(new BigDecimal(amountToReceive.bigIntegerValue())
+          .divide(senderUnitsPerReceiverUnits, RoundingMode.FLOOR).toBigInteger()); // FIXME how to round
+    }
+
+    @Override
+    public UnsignedLong calculateMinAmountToAccept(UnsignedLong sendAmount,
+        Denomination sendDenomination,
+        Optional<Denomination> expectedReceivedDenomination) {
+      return UnsignedLong.valueOf(new BigDecimal(sendAmount.bigIntegerValue())
+          .multiply(senderUnitsPerReceiverUnits).toBigInteger()); // FIXME how to round
+    }
+  }
+
 }
diff --git a/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/testutils/SimulatedIlpv4Network.java b/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/testutils/SimulatedIlpv4Network.java
index ba4abcd..d5b7be4 100644
--- a/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/testutils/SimulatedIlpv4Network.java
+++ b/stream-parent/stream-receiver/src/test/java/org/interledger/stream/receiver/testutils/SimulatedIlpv4Network.java
@@ -1,5 +1,7 @@
 package org.interledger.stream.receiver.testutils;
 
+import static org.interledger.stream.FluentCompareTo.is;
+
 import org.interledger.core.InterledgerAddress;
 import org.interledger.core.InterledgerErrorCode;
 import org.interledger.core.InterledgerPreparePacket;
@@ -162,7 +164,7 @@ public class SimulatedIlpv4Network {
     Objects.requireNonNull(multiplier);
 
     BigInteger newAmount = multiplier.multiply(new BigDecimal(preparePacket.getAmount().bigIntegerValue()))
-        .toBigIntegerExact();
+        .toBigInteger();
     return InterledgerPreparePacket.builder().from(preparePacket).amount(UnsignedLong.valueOf(newAmount)).build();
   }
 
@@ -210,6 +212,6 @@ public class SimulatedIlpv4Network {
   ) {
     Objects.requireNonNull(simulatedPathConditions);
     Objects.requireNonNull(preparePacket);
-    return preparePacket.getAmount().compareTo(simulatedPathConditions.maxPacketAmount().get()) > 0;
+    return is(preparePacket.getAmount()).greaterThan(simulatedPathConditions.maxPacketAmount().get());
   }
 }