Make Joystream protocole_session compatible with streaming

include/protocol_session/Exceptions.hpp

@@ -177,6 +177,15 @@ class PeerNotReadyToStartUploading : public std::runtime_error {
     PeerNotReadyToStartUploadingCause peerNotReadyToStartUploadingCause;
 };
 
+class NoPieceAvailableException : public std::runtime_error {
+
+public:
+
+    NoPieceAvailableException()
+        : std::runtime_error(std::string("No more piece available")) {
+    }
+};
+
 }
 }
 }

include/protocol_session/Session.cpp

@@ -478,7 +478,9 @@ namespace protocol_session {
     }
 
     template<class ConnectionIdType>
-    void Session<ConnectionIdType>::startDownloading(const Coin::Transaction & contractTx, const PeerToStartDownloadInformationMap<ConnectionIdType> & peerToStartDownloadInformationMap) {
+    void Session<ConnectionIdType>::startDownloading(const Coin::Transaction & contractTx,
+                                                     const PeerToStartDownloadInformationMap<ConnectionIdType> & peerToStartDownloadInformationMap,
+                                                     const PickNextPieceMethod<ConnectionIdType> & pickNextPieceMethod) {
 
         if(_state == SessionState::paused)
             throw exception::StateIncompatibleOperation("cannot start downloading on a paused session.");
@@ -499,7 +501,7 @@ namespace protocol_session {
             case SessionMode::buying:
 
                 assert(_observing == nullptr && _buying != nullptr && _selling == nullptr);
-
+                _buying->setPickNextPieceMethod(pickNextPieceMethod);
                 _buying->startDownloading(contractTx, peerToStartDownloadInformationMap);
                 break;
 
@@ -1015,4 +1017,3 @@ namespace protocol_session {
     }
 }
 }
-

include/protocol_session/Session.hpp

@@ -9,6 +9,7 @@
 #define JOYSTREAM_PROTOCOLSESSION_SESSION_HPP
 
 #include <protocol_session/detail/Connection.hpp>
+#include <protocol_session/detail/Piece.hpp>
 #include <protocol_session/Callbacks.hpp>
 #include <protocol_session/SessionMode.hpp>
 #include <protocol_session/SessionState.hpp>
@@ -36,6 +37,9 @@ namespace detail {
     class Observing;
 }
 
+    template <class ConnectionIdType>
+    using PickNextPieceMethod = std::function<int(const std::vector<detail::Piece<ConnectionIdType>>*)>;
+
     class TorrentPieceInformation;
 
     template <class ConnectionIdType>
@@ -147,7 +151,8 @@ namespace detail {
          * @throws exception::NoLongerSendingInvitations if invitations are no longer being sent in buying mode (_state != BuyingState::sending_invitations)
          */
         void startDownloading(const Coin::Transaction & contractTx,
-                              const PeerToStartDownloadInformationMap<ConnectionIdType> & peerToStartDownloadInformationMap);
+                              const PeerToStartDownloadInformationMap<ConnectionIdType> & peerToStartDownloadInformationMap,
+                              const PickNextPieceMethod<ConnectionIdType> & pickNextPieceMethod);
 
         // A valid piece was sent too us on given connection
         void validPieceReceivedOnConnection(const ConnectionIdType &, int index);
@@ -275,4 +280,3 @@ namespace detail {
 #include <protocol_session/Session.cpp>
 
 #endif // JOYSTREAM_PROTOCOLSESSION_SESSION_HPP
-

include/protocol_session/detail/Buying.cpp

@@ -575,9 +575,8 @@ namespace detail {
             pieceIndex = _deAssignedPieces.front();
             _deAssignedPieces.pop_front();
         } else {
-
             try {
-                pieceIndex = getNextUnassignedPiece();
+                pieceIndex = this->_pickNextPieceMethod(&_pieces);
             } catch(const std::runtime_error & e) {
                 // No unassigned piece was found
                 return false;
@@ -596,23 +595,6 @@ namespace detail {
         return true;
     }
 
-    template <class ConnectionIdType>
-    int Buying<ConnectionIdType>::getNextUnassignedPiece() const {
-
-         // Look in interval [_assignmentLowerBound, end]
-         for(uint32_t i = _assignmentLowerBound;i < _pieces.size();i++)
-             if(_pieces[i].state() == PieceState::unassigned)
-                 return i;
-
-         // Then try interval [0, _assignmentLowerBound)
-         for(uint32_t i = 0;i < _assignmentLowerBound;i++)
-             if(_pieces[i].state() == PieceState::unassigned)
-                 return i;
-
-         // We did not find anything
-         throw std::runtime_error("Unable to find any unassigned pieces.");
-    }
-
     template<class ConnectionIdType>
     typename detail::ConnectionMap<ConnectionIdType>::const_iterator Buying<ConnectionIdType>::removeConnection(const ConnectionIdType & id, DisconnectCause cause) {
 
@@ -701,6 +683,11 @@ namespace detail {
         return _terms;
     }
 
+    template <class ConnectionIdType>
+    void Buying<ConnectionIdType>::setPickNextPieceMethod(const PickNextPieceMethod<ConnectionIdType> & pickNextPieceMethod) {
+      _pickNextPieceMethod = pickNextPieceMethod;
+    }
+
 }
 }
 }

include/protocol_session/detail/Buying.hpp

@@ -112,16 +112,15 @@ class Buying {
 
     protocol_wire::BuyerTerms terms() const;
 
+    void setPickNextPieceMethod(const PickNextPieceMethod<ConnectionIdType> & pickNextPieceMethod);
+
 private:
 
     //// Assigning pieces
 
     // Tries to assign an unassigned piece to given seller
     bool tryToAssignAndRequestPiece(detail::Seller<ConnectionIdType> &);
 
-    // Tries to find next unassigned piece
-    int getNextUnassignedPiece() const;
-
     //// Utility routines
 
     // Prepare given connection for deletion due to given cause
@@ -186,6 +185,9 @@ class Buying {
     // (i.e. entered state StartedState::sending_invitations).
     // Is used to figure out when to start trying to build the contract
     std::chrono::high_resolution_clock::time_point _lastStartOfSendingInvitations;
+
+    // Function that if defined will return the next piece that we should download
+    PickNextPieceMethod<ConnectionIdType> _pickNextPieceMethod;
 };
 
 }