Repository: hyperledger/quilt
Pull request #464 (2020-05-14 20:50:20)

send stream close frame when all is said and done on a send money request


Patch:
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 232e707..ddf204d 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
@@ -32,6 +32,8 @@ import org.interledger.stream.crypto.JavaxStreamEncryptionService;
 import org.interledger.stream.crypto.StreamEncryptionService;
 import org.interledger.stream.frames.ConnectionAssetDetailsFrame;
 import org.interledger.stream.frames.ConnectionNewAddressFrame;
+import org.interledger.stream.frames.ErrorCodes;
+import org.interledger.stream.frames.StreamCloseFrame;
 import org.interledger.stream.frames.StreamFrame;
 import org.interledger.stream.frames.StreamFrameType;
 import org.interledger.stream.frames.StreamMoneyFrame;
@@ -50,6 +52,7 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -66,7 +69,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.function.Supplier;
-
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -382,6 +384,11 @@ public class SimpleStreamSender implements StreamSender {
           }, sendMoneyExecutor)
           .whenComplete(($, error) -> {
             sendMoneyExecutor.shutdown();
+            try {
+              closeStream();
+            } catch (StreamConnectionClosedException e) {
+              logger.error("closeStream failed", e);
+            }
             if (error != null) {
               logger.error("SendMoney Stream failed: " + error.getMessage(), error);
             }
@@ -964,27 +971,26 @@ public class SimpleStreamSender implements StreamSender {
       this.unrecoverableErrorEncountered.set(unrecoverableErrorEncountered);
     }
 
+    /**
+     * Close the current STREAM connection by sending a {@link StreamCloseFrame} to the receiver.
+     *
+     * @return An {@link UnsignedLong} representing the amount delivered by this individual stream.
+     */
+     @VisibleForTesting
+     void closeStream() throws StreamConnectionClosedException {
+       this.sendStreamFramesInZeroValuePacket(Lists.newArrayList(
+           StreamCloseFrame.builder()
+               .streamId(UnsignedLong.ONE)
+               .errorCode(ErrorCodes.NoError)
+               .build()
+       ));
+     }
+
     // TODO: FIXME per https://github.com/hyperledger/quilt/issues/308. Until this is completed, parallel STREAM send()
     //  operations are not thread-safe. Consider a new instance of everything on each sendMoney? In other words, each
     //  SendMoney requires its own connection information,
     // so as long as two senders don't use the same Connection details, everything should be thread-safe.
 
-    ///**
-    // * Close the current STREAM connection by sending a {@link ConnectionCloseFrame} to the receiver.
-    // *
-    // * @return An {@link UnsignedLong} representing the amount delivered by this individual stream.
-    // */
-    // TODO: Add unit test coverage here per See https://github.com/hyperledger/quilt/issues/308
-    // @VisibleForTesting
-    // void closeStream() throws StreamConnectionClosedException {
-    //   this.sendStreamFramesInZeroValuePacket(Lists.newArrayList(
-    //       StreamCloseFrame.builder()
-    //           .streamId(UnsignedLong.ONE)
-    //           .errorCode(ErrorCode.NoError)
-    //           .build()
-    //   ));
-    // }
-
     // /**
     //  * Close the current STREAM connection by sending a {@link ConnectionCloseFrame} to the receiver.
     //  *
@@ -992,72 +998,68 @@ public class SimpleStreamSender implements StreamSender {
     //  */
     // // TODO: Add unit test coverage here per See https://github.com/hyperledger/quilt/issues/308
     // @VisibleForTesting
-    // void closeConnection() throws StreamConnectionClosedException {
-    //   this.sendStreamFramesInZeroValuePacket(Lists.newArrayList(
-    //       ConnectionCloseFrame.builder()
-    //           .errorCode(ErrorCode.NoError)
-    //           .build()
-    //   ));
-    // }
-
-    // private void sendStreamFramesInZeroValuePacket(final Collection<StreamFrame> streamFrames)
-    //     throws StreamConnectionClosedException {
-    //   Objects.requireNonNull(streamFrames);
-    //
-    //   if (streamFrames.size() <= 0) {
-    //     logger.warn("sendStreamFrames called with 0 frames");
-    //     return;
-    //   }
-    //
-    //   final StreamPacket streamPacket = StreamPacket.builder()
-    //       .interledgerPacketType(InterledgerPacketType.PREPARE)
-    //       .prepareAmount(UnsignedLong.ZERO)
-    //       .sequence(streamConnection.nextSequence())
-    //       .addAllFrames(streamFrames)
-    //       .build();
-    //
-    //   // Create the ILP Prepare packet using an encrypted StreamPacket as the encryptedStreamPacket payload...
-    //   final byte[] encryptedStreamPacket = this.toEncrypted(sharedSecret, streamPacket);
-    //   final InterledgerCondition executionCondition;
-    //   executionCondition = generatedFulfillableFulfillment(sharedSecret, encryptedStreamPacket).getCondition();
-    //
-    //   final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
-    //       .destination(destinationAddress)
-    //       .amount(UnsignedLong.ZERO)
-    //       .executionCondition(executionCondition)
-    //       .expiresAt(TimeUtils.now().plusSeconds(30L))
-    //       .data(encryptedStreamPacket)
-    //       .build();
-    //
-    //   final PrepareAmounts prepareAmounts = PrepareAmounts.from(preparePacket, streamPacket);
-    //
-    //   link.sendPacket(preparePacket).handle(
-    //       fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket, prepareAmounts),
-    //       rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, prepareAmounts, numRejectedPackets,
-    //           congestionController)
-    //   );
-    //
-    //   // Mark the streamConnection object as closed if the caller supplied a ConnectionCloseFrame
-    //   streamFrames.stream()
-    //       .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionClose)
-    //       .findAny()
-    //       .ifPresent($ -> {
-    //         streamConnection.closeConnection();
-    //         logger.info("STREAM Connection closed.");
-    //       });
-    //
-    //   // Emit a log statement if the called supplied a StreamCloseFrame
-    //   streamFrames.stream()
-    //       .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamClose)
-    //       .findAny()
-    //       .map($ -> (StreamCloseFrame) $)
-    //       .ifPresent($ -> {
-    //         logger.info(
-    //             "StreamId {} Closed. Delivered: {} ({} packets fulfilled, {} packets rejected)",
-    //             $.streamId(), paymentTracker.getDeliveredAmountInReceieverUnits(), this.numFulfilledPackets.get(),
-    //             this.numRejectedPackets.get()
-    //         );
-    //       });
-    // }
+//     void closeConnection() throws StreamConnectionClosedException {
+//       this.sendStreamFramesInZeroValuePacket(Lists.newArrayList(
+//           ConnectionCloseFrame.builder()
+//               .errorCode(ErrorCodes.NoError)
+//               .build()
+//       ));
+//     }
+
+     private void sendStreamFramesInZeroValuePacket(final Collection<StreamFrame> streamFrames)
+         throws StreamConnectionClosedException {
+       Objects.requireNonNull(streamFrames);
+
+       if (streamFrames.size() <= 0) {
+         logger.warn("sendStreamFrames called with 0 frames");
+         return;
+       }
+
+       final StreamPacket streamPacket = StreamPacket.builder()
+           .interledgerPacketType(InterledgerPacketType.PREPARE)
+           .prepareAmount(UnsignedLong.ZERO)
+           .sequence(streamConnection.nextSequence())
+           .addAllFrames(streamFrames)
+           .build();
+
+       // Create the ILP Prepare packet using an encrypted StreamPacket as the encryptedStreamPacket payload...
+       final byte[] encryptedStreamPacket = this.toEncrypted(sharedSecret, streamPacket);
+       final InterledgerCondition executionCondition;
+       executionCondition = generatedFulfillableFulfillment(sharedSecret, encryptedStreamPacket).getCondition();
+
+       final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
+           .destination(destinationAddress)
+           .amount(UnsignedLong.ZERO)
+           .executionCondition(executionCondition)
+           .expiresAt(DateUtils.now().plusSeconds(30L))
+           .data(encryptedStreamPacket)
+           .typedData(streamPacket)
+           .build();
+
+       link.sendPacket(preparePacket);
+
+       //TODO: FIXME per https://github.com/hyperledger/quilt/issues/308
+       // Mark the streamConnection object as closed if the caller supplied a ConnectionCloseFrame
+//       streamFrames.stream()
+//           .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionClose)
+//           .findAny()
+//           .ifPresent($ -> {
+//             streamConnection.closeConnection();
+//             logger.info("STREAM Connection closed.");
+//           });
+
+       // Emit a log statement if the called supplied a StreamCloseFrame
+       streamFrames.stream()
+           .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamClose)
+           .findAny()
+           .map($ -> (StreamCloseFrame) $)
+           .ifPresent($ -> {
+             logger.info(
+                 "StreamId {} Closed. Delivered: {} ({} packets fulfilled, {} packets rejected)",
+                 $.streamId(), paymentTracker.getDeliveredAmountInReceiverUnits(), this.numFulfilledPackets.get(),
+                 this.numRejectedPackets.get()
+             );
+           });
+     }
   }
 }
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 97225da..930893f 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
@@ -24,6 +24,7 @@ import static org.interledger.stream.FluentCompareTo.is;
 
 import org.interledger.core.Immutable;
 import org.interledger.core.InterledgerPacketType;
+import org.interledger.core.SharedSecret;
 import org.interledger.stream.frames.StreamFrame;
 import org.interledger.stream.frames.StreamFrameType;
 
@@ -31,6 +32,7 @@ import com.google.common.primitives.UnsignedLong;
 import org.immutables.value.Value.Derived;
 
 import java.util.List;
+import java.util.Optional;
 
 /**
  * <p>Interledger Payments moves assets of one party to another that consists of one or more ledger
@@ -135,4 +137,10 @@ public interface StreamPacket {
    */
   List<StreamFrame> frames();
 
+  /**
+   * transient value of shared secret optionally set by local receiver but not encoded across the wire
+   * @return
+   */
+  Optional<SharedSecret> sharedSecret();
+
 }
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 5a9a1c8..b3e4a91 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
@@ -156,6 +156,7 @@ public class StatelessStreamReceiver implements StreamReceiver {
           .interledgerPacketType(InterledgerPacketType.FULFILL)
           .prepareAmount(preparePacket.getAmount())
           .frames(responseFrames.build())
+          .sharedSecret(streamSharedSecret)
           .build();
 
       logger.debug(


Test patch:
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
index a7a6e1f..aedf384 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
@@ -23,6 +23,7 @@ import static org.interledger.core.InterledgerErrorCode.T04_INSUFFICIENT_LIQUIDI
 import static org.interledger.core.InterledgerErrorCode.T05_RATE_LIMITED;
 import static org.interledger.core.InterledgerErrorCode.T99_APPLICATION_ERROR;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.atMost;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -56,6 +57,9 @@ import org.interledger.stream.StreamConnectionId;
 import org.interledger.stream.StreamPacket;
 import org.interledger.stream.calculators.NoOpExchangeRateCalculator;
 import org.interledger.stream.crypto.StreamEncryptionService;
+import org.interledger.stream.frames.ErrorCodes;
+import org.interledger.stream.frames.StreamCloseFrame;
+import org.interledger.stream.frames.StreamFrame;
 import org.interledger.stream.frames.StreamMoneyFrame;
 import org.interledger.stream.sender.SimpleStreamSender.SendMoneyAggregator;
 
@@ -75,6 +79,7 @@ import org.mockito.stubbing.Answer;
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -84,6 +89,7 @@ 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.stream.Collectors;
 
 /**
  * Unit tests for {@link SendMoneyAggregator}.
@@ -152,8 +158,7 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 1 Link call due to preflight check
-    Mockito.verify(linkMock, times(1)).sendPacket(any());
+    assertOnlyPreflightAndClose();
   }
 
   @Test
@@ -165,8 +170,7 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 1 Link call due to preflight check
-    Mockito.verify(linkMock, times(1)).sendPacket(any());
+    assertOnlyPreflightAndClose();
   }
 
   @Test
@@ -177,9 +181,7 @@ public class SendMoneyAggregatorTest {
     when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
 
     sendMoneyAggregator.send().get();
-
-    // Expect 1 Link call due to preflight check
-    Mockito.verify(linkMock, times(1)).sendPacket(any());
+    assertOnlyPreflightAndClose();
   }
 
   @Test
@@ -192,8 +194,7 @@ public class SendMoneyAggregatorTest {
 
     sendMoneyAggregator.send().get();
 
-    // Expect 1 Link call due to preflight check
-    Mockito.verify(linkMock, times(1)).sendPacket(any());
+    assertOnlyPreflightAndClose();
   }
 
   @Test
@@ -656,6 +657,7 @@ public class SendMoneyAggregatorTest {
     assertThat(result.amountLeftToSend()).isLessThanOrEqualTo(UnsignedLong.valueOf(9));
     verify(congestionControllerMock, atMost(11)).getMaxAmount();
     verify(linkMock, atMost(11)).sendPacket(any());
+    assertStreamCloseFrameSent();
   }
 
   /**
@@ -747,4 +749,38 @@ public class SendMoneyAggregatorTest {
         .build();
   }
 
+  /**
+   * Extracts all the stream frames from all the prepare packets sent on the link.
+   * @return
+   */
+  private List<StreamFrame> getStreamFramesSent() {
+    ArgumentCaptor<InterledgerPreparePacket> prepareCaptor = ArgumentCaptor.forClass(InterledgerPreparePacket.class);
+    verify(linkMock, atLeastOnce()).sendPacket(prepareCaptor.capture());
+    return prepareCaptor.getAllValues().stream()
+        .map(InterledgerPreparePacket::typedData)
+        .filter(Optional::isPresent)
+        .map(Optional::get)
+        .filter($ -> $ instanceof StreamPacket)
+        .map($ -> (StreamPacket) $)
+        .flatMap(packet -> packet.frames().stream())
+        .collect(Collectors.toList());
+  }
+
+  /**
+   * Assert that only preflight and close frames were sent, as is the case when no other frames can/need to be sent.
+   */
+  private void assertOnlyPreflightAndClose() {
+    // Expect 2 Link calls due to preflight check and stream close
+    Mockito.verify(linkMock, times(2)).sendPacket(any());
+    assertStreamCloseFrameSent();
+  }
+
+  private void assertStreamCloseFrameSent() {
+    assertThat(getStreamFramesSent()).contains(closeFrame());
+  }
+
+  private StreamCloseFrame closeFrame() {
+    return StreamCloseFrame.builder().streamId(UnsignedLong.ONE).errorCode(ErrorCodes.NoError).build();
+  }
+
 }
diff --git a/stream-parent/stream-core/src/test/java/org/interledger/stream/StreamPacketTest.java b/stream-parent/stream-core/src/test/java/org/interledger/stream/StreamPacketTest.java
index 9b6df4c..c919c58 100644
--- a/stream-parent/stream-core/src/test/java/org/interledger/stream/StreamPacketTest.java
+++ b/stream-parent/stream-core/src/test/java/org/interledger/stream/StreamPacketTest.java
@@ -3,12 +3,14 @@ package org.interledger.stream;
 import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
 
 import org.interledger.core.InterledgerPacketType;
+import org.interledger.core.SharedSecret;
 import org.interledger.stream.frames.StreamFrame;
 
 import com.google.common.primitives.UnsignedLong;
 import org.junit.Test;
 
 import java.util.List;
+import java.util.Optional;
 
 /**
  * Unit tests for {@link StreamPacket}.
@@ -44,6 +46,11 @@ public class StreamPacketTest {
       public List<StreamFrame> frames() {
         return null;
       }
+
+      @Override
+      public Optional<SharedSecret> sharedSecret() {
+        return Optional.empty();
+      }
     };
     assertThat(interfacePacket.version()).isEqualTo((short) 1);
   }