(#12) Add ipipe variables for first data exchange

move/Move.lock

@@ -23,6 +23,6 @@ dependencies = [
 ]
 
 [move.toolchain-version]
-compiler-version = "1.18.1"
+compiler-version = "1.18.0"
 edition = "legacy"
 flavor = "sui"

move/sources/errors.move

@@ -16,6 +16,7 @@ module dtp::errors {
     public fun EServiceIdxOutOfRange() : u64 { 5 }
     public fun EInvalidAccessOnNone() : u64 { 6 }
     public fun EHostNotOwner() : u64 { 7 }
+    public fun EInvalidPipeCount() : u64 { 8 }
 
   // === Structs ===
 

move/sources/events.move

@@ -14,9 +14,20 @@ module dtp::events {
 
   // === Structs ===
     struct ConReq has copy, drop {
-        // TODO Add ServiceTYpe, Pipe and InnerPipe addresses.
-        tc_address: address, // Transport Control Address.
-        sender: address, // Sender requesting the connection.
+        service_idx: u8, // Service Type
+        cli_haddr: address, // Client Host address (requester of the connection).
+        srv_haddr: address, // Server Host address
+        tc_addr: address, // Transport Control        
+
+        // Address of the first inner pipe addresses to use.
+        //
+        // This is for faster initial response time
+        // for some services (e.g. first ping).
+        //
+        // The server should find out about additional
+        // inner pipes with a tc_addr object read.
+        client_tx_ipipe: address,
+        server_tx_ipipe: address,
     }
 
 
@@ -27,8 +38,8 @@ module dtp::events {
   // === Admin Functions ===
 
   // === Public-Friend Functions ===
-  public(friend) fun emit_con_req( tc_address: address, sender: address ) {
-    event::emit(ConReq {tc_address, sender} );
+  public(friend) fun emit_con_req( service_idx: u8, cli_haddr: address, srv_haddr: address, tc_addr: address, client_tx_ipipe: address, server_tx_ipipe: address ) {
+    event::emit(ConReq { service_idx, cli_haddr, srv_haddr, tc_addr, client_tx_ipipe, server_tx_ipipe});
   }
 
   // === Private Functions ===

move/sources/host.move

@@ -128,6 +128,7 @@ module dtp::host {
         }
     }
 
+    #[allow(lint(share_owned))]
     public(friend) fun new_transfered( creator: address, ctx: &mut TxContext ): WeakRef 
     {
       let new_obj = new(creator,ctx);

move/sources/inner_pipe.move

@@ -52,17 +52,17 @@ module dtp::inner_pipe {
 
   // === Public-Friend Functions ===
 
-    public(friend) fun new( pipe_address: address, ctx: &mut TxContext  ): InnerPipe {
+    public(friend) fun new( pipe_address: &address, ctx: &mut TxContext  ): InnerPipe {
         let new_obj = InnerPipe {
           id: object::new(ctx),
           flgs: 0u8,
-          pipe_id: weak_ref::new_from_address(pipe_address),
+          pipe_id: weak_ref::new_from_address(*pipe_address),
           sync_data: pipe_sync_data::new(),
         };
         new_obj
     }
 
-    public(friend) fun new_transfered( pipe_address: address, recipient: address, ctx: &mut TxContext  ): WeakRef 
+    public(friend) fun new_transfered( pipe_address: &address, recipient: address, ctx: &mut TxContext  ): WeakRef 
     {
       let new_obj = new(pipe_address,ctx);
       let new_obj_ref = weak_ref::new_from_address(uid_to_address(&new_obj.id));

move/sources/pipe.move

@@ -13,9 +13,9 @@ module dtp::pipe {
     use sui::transfer::{Self};
     use sui::tx_context::{TxContext};
     use dtp::weak_ref::{Self,WeakRef};
-    //use dtp::errors::{Self};
-    //use dtp::transport_control;
+    use dtp::errors::{Self};    
     use dtp::pipe_sync_data::{Self,PipeSyncData};
+    use dtp::inner_pipe::{Self};
 
   // === Friends ===
     friend dtp::host;
@@ -40,13 +40,6 @@ module dtp::pipe {
         inner_pipes: vector<WeakRef>,
     }
 
-    struct InnerPipe has key, store {
-        id: UID,
-        flgs: u8, // DTP version+esc flags always after UID.
-
-        pipe_id: WeakRef,
-    }
-
   // === Public-Mutative Functions ===
 
   // === Public-View Functions ===
@@ -55,35 +48,53 @@ module dtp::pipe {
 
   // === Public-Friend Functions ===
 
-    public(friend) fun new_transfered( tctl_id: &ID, inner_pipe_count: u8, recipient: address, ctx: &mut TxContext ): WeakRef {
+    public(friend) fun new_transfered( tctl_id: &ID, inner_pipe_count: u8, recipient: address, ctx: &mut TxContext ): (WeakRef,WeakRef) {
+        assert!(inner_pipe_count > 0, errors::EInvalidPipeCount());
+
         let new_pipe = Pipe {
             id: object::new(ctx),
             flgs: 0,            
             sync_data: pipe_sync_data::new(),
             tctl_id: weak_ref::new(tctl_id),
             inner_pipes: vector::empty(),
         };
-        let new_pipe_ref = weak_ref::new_from_address(uid_to_address(&new_pipe.id));
+        let pipe_address = uid_to_address(&new_pipe.id);
+
+        // First InnerPipe created is "special" because the caller gets a WeakRef on it.
+        let ipipe_ref = inner_pipe::new_transfered(&pipe_address, recipient, ctx);
+        let ipipe_addr = weak_ref::get_address(&ipipe_ref);
+        vector::push_back(&mut new_pipe.inner_pipes, ipipe_ref);
+
+        // Create additional InnerPipes.
+        inner_pipe_count = inner_pipe_count - 1;
         let i: u8 = 0;
         while (i < inner_pipe_count) {
-            let inner_pipe = InnerPipe {
-                id: object::new(ctx),
-                flgs: 0u8,
-                pipe_id: new_pipe_ref,
-            };
-            let inner_pipe_ref = weak_ref::new_from_address(uid_to_address(&inner_pipe.id));
-            vector::push_back(&mut new_pipe.inner_pipes, inner_pipe_ref);
-            transfer::transfer(inner_pipe, recipient);
+            let ipipe_ref = inner_pipe::new_transfered(&pipe_address, recipient, ctx);
+            vector::push_back(&mut new_pipe.inner_pipes, ipipe_ref);
         };
-        
+
         transfer::transfer(new_pipe, recipient);
-        new_pipe_ref
+
+        (weak_ref::new_from_address(pipe_address), weak_ref::new_from_address(ipipe_addr))
     }
 
-    public(friend) fun delete( self: Pipe ) {
-        let Pipe { id, flgs: _, sync_data: _, tctl_id: _, inner_pipes: _ } = self;        
+/* TODO
+    public(friend) fun delete( self: Pipe, inner_pipes: vector<InnerPipe> ) {
+        let Pipe { id, flgs: _, sync_data: _, tctl_id: _, inner_pipes } = self;        
+        // Delete all the inner pipes.
+        // For tracking/debugging purpose, the weak ref is not removed from vector (only cleared).
+        let i: u64 = 0;
+        while (i < vector::length(&inner_pipes)) {
+            let inner_pipe_ref = vector::borrow_mut(&mut inner_pipes, i);
+            let inner_pipe_id = weak_ref::id(inner_pipe_ref);
+            weak_ref::clear(inner_pipe_ref);
+            object::delete(inner_pipe_id);
+            inner_pipe::delete(inner_pipe_id);
+            i = i + 1;
+        };
         object::delete(id);
     }
+*/
 
   // === Private Functions ===
 

move/sources/transport_control.move

@@ -69,18 +69,27 @@ module dtp::transport_control {
         // Intended for slow discovery.
         //
         // It is expected that DTP off-chain will cache these IDs.
-        //
-        // Optional in case of uni-directiobal connection.
         client_tx_pipe: WeakRef,
         server_tx_pipe: WeakRef,
+
+        // Inner Pipe to be used for first request/response.
+        //
+        // This is an optimization for faster first interactions
+        // after connection creation.
+        //
+        // The endpoints can get all the inner pipe addresses with
+        // read of the client_tx_pipe and server_tx_pipe objects.
+        client_tx_ipipe: WeakRef,
+        server_tx_ipipe: WeakRef,
+
     }
 
 
     // Constructors
 
     public(friend) fun new( service_idx: u8,
         client_host: &mut Host,
-        server_host: &mut Host,
+        server_host: &Host,
         ctx: &mut TxContext): TransportControl
     {
         // Check service_idx is in-range.        
@@ -119,11 +128,17 @@ module dtp::transport_control {
             server_addr: host::creator(server_host),
             client_tx_pipe: weak_ref::new_empty(),
             server_tx_pipe: weak_ref::new_empty(),
+            client_tx_ipipe: weak_ref::new_empty(),
+            server_tx_ipipe: weak_ref::new_empty(),
         };
 
-        // Weak references between Pipes and TC (for recovery scenario).        
-        tc.client_tx_pipe = dtp::pipe::new_transfered(object::borrow_id<TransportControl>(&tc), 2, tc.client_addr, ctx);
-        tc.server_tx_pipe = dtp::pipe::new_transfered(object::borrow_id<TransportControl>(&tc), 2, tc.server_addr, ctx);
+        // Initialize all the Weak references (for slow discovery).
+        let (a,b) = dtp::pipe::new_transfered(object::borrow_id<TransportControl>(&tc), 2, tc.client_addr, ctx);
+        tc.client_tx_pipe = a;
+        tc.client_tx_ipipe= b;
+        (a,b) = dtp::pipe::new_transfered(object::borrow_id<TransportControl>(&tc), 2, tc.server_addr, ctx);
+        tc.server_tx_pipe = a;
+        tc.server_tx_ipipe= b;
 
         tc
     }
@@ -135,8 +150,10 @@ module dtp::transport_control {
           client_host: _, server_host: _,
           client_addr: _, server_addr: _,
           client_tx_pipe: _, server_tx_pipe: _,
+          client_tx_ipipe: _, server_tx_ipipe: _
         } = self;
 
+
         object::delete(id);
     }    
 
@@ -149,6 +166,12 @@ module dtp::transport_control {
         self.client_addr
     }
 
+    // Initial Inner Pipe address.
+    // This is for the first response toward the client.
+    public(friend) fun ipipe_addr(self: &TransportControl): address {
+        weak_ref::get_address(&self.server_tx_pipe)
+    }
+
     // The TransportControl is the shared object for the
     // connection between two hosts.
     //
@@ -173,19 +196,24 @@ module dtp::transport_control {
     // the expectation realistic (base on recent server stats) that the service will be
     // available.
     //
-    // Best-effort has very little control and could be decline in case of a server
-    // being too busy. At worst, the server is not going to respond to the connection 
-    // request and the client will have wasted gas attempting to connect.
+    // Best-effort has very little control and could fail or be decline in case of 
+    // a server being too busy. At worst, the server is not going to respond to the 
+    // connection request and the client will have wasted gas attempting to connect.
     //
     // Pre-approved Service is a more controlled approach where the connection is
-    // "guaranteed" to work for the approved user. It is more appropriate in
-    // circumstance where there is an off-chain business relationship.
+    // "guaranteed" to have dedicated ressources (no oversubscribing). It is more 
+    // appropriate in circumstance where there is an off-chain business relationship.
     //
+    // Returns:
+    //    - tc_address: TransportControl address. Most other addresses can be learn from it.
+    //    - client_tx_ipipe: First InnerPipe used by client to TX to server.
+    //    - server_tx_ipipe: First InnerPipe used by server to TX to client.
+
     #[allow(lint(share_owned))]
     public entry fun create_best_effort( service_idx: u8,
                                          client_host: &mut Host,
-                                         server_host: &mut Host,                                         
-                                         ctx: &mut TxContext )    
+                                         server_host: &Host,
+                                         ctx: &mut TxContext ): (address,address,address)
     {
         // Sender must be the owner of the client_host.    
         assert!(host::is_caller_creator(client_host, ctx), errors::EHostNotOwner());
@@ -202,8 +230,16 @@ module dtp::transport_control {
         // Emit the "Connection Request" Move event.
         // The server will see the sender address therefore will know the TC and plenty of info!
         let tc_address = object::id_to_address( object::borrow_id<TransportControl>(&tc) );
-        dtp::events::emit_con_req( tc_address, client_address(&tc) );
+        dtp::events::emit_con_req( service_idx,  
+                        client_address(&tc), server_address(&tc), 
+                        tc_address, client_tx_ipipe(&tc), server_tx_ipipe(&tc));
+        let client_tx_ipipe = client_tx_ipipe(&tc);
+        let server_tx_ipipe = server_tx_ipipe(&tc);
         transfer::share_object(tc);
+
+        // TODO Add the connection to the Client Host object registry (for slow discovery).
+
+        (tc_address, client_tx_ipipe, server_tx_ipipe)
     }
 
     public entry fun create_preapproved( _ctx: &mut TxContext ) 
@@ -219,6 +255,14 @@ module dtp::transport_control {
         self.server_addr
     }
 
+    public(friend) fun client_tx_ipipe(self: &TransportControl): address {        
+        weak_ref::get_address(&self.client_tx_ipipe)
+    }
+
+    public(friend) fun server_tx_ipipe(self: &TransportControl): address {
+        weak_ref::get_address(&self.server_tx_ipipe)
+    }
+
     // Connection State Machine (work-in-progress):
     //
     //  Healthy transitions:
@@ -250,23 +294,23 @@ module dtp::transport_control {
 #[test_only]
 module dtp::test_transport_control {
     //use std::debug;
-    use std::option::{Self};
+    //use std::option::{Self};
 
-    use sui::transfer;
+    //use sui::transfer;
     use sui::test_scenario::{Scenario};
     use sui::test_scenario as ts;
-    use sui::object;
+    //use sui::object;
 
-    use dtp::pipe::{Self};
-    use dtp::transport_control::{Self};  // DUT
+    //use dtp::pipe::{Self};
+    //use dtp::transport_control::{Self};  // DUT
     use dtp::host::{Self};
 
     fun create_hosts(scenario: &mut Scenario) {    
         ts::next_tx(scenario, @0x10);
         {
             let sender = ts::sender(scenario);
             let ctx = ts::ctx(scenario);
-            let client_host = host::new_transfered(sender, ctx);
+            let _client_host = host::new_transfered(sender, ctx);
         };
 
         ts::next_tx(scenario, @0x20);

move/sources/weak_ref.move

@@ -156,7 +156,7 @@ module dtp::weak_ref {
         assert!(is_set(self), errors::EInvalidAccessOnNone());
         self.reference
     }
-
+    
   // === Private Functions ===
 
   // === Test Functions ===