+193 -171

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go

··· 399 399 } 400 400 { 401 401 checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 402 402 + return C.uniffi_iroh_streamplace_checksum_method_db_shutdown() 403 403 + }) 404 404 + if checksum != 9825 { 405 405 + // If this happens try cleaning and rebuilding your project 406 406 + panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_db_shutdown: UniFFI API checksum mismatch") 407 407 + } 408 408 + } 409 409 + { 410 410 + checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 402 411 return C.uniffi_iroh_streamplace_checksum_method_db_subscribe() 403 412 }) 404 413 if checksum != 415 { ··· 548 557 if checksum != 18989 { 549 558 // If this happens try cleaning and rebuilding your project 550 559 panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_node_send_segment: UniFFI API checksum mismatch") 560 560 + } 561 561 + } 562 562 + { 563 563 + checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 564 564 + return C.uniffi_iroh_streamplace_checksum_method_node_shutdown() 565 565 + }) 566 566 + if checksum != 18129 { 567 567 + // If this happens try cleaning and rebuilding your project 568 568 + panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_node_shutdown: UniFFI API checksum mismatch") 551 569 } 552 570 } 553 571 { ··· 1172 1190 // Iroh-streamplace specific metadata database. 1173 1191 type DbInterface interface { 1174 1192 IterWithOpts(filter *Filter) ([]Entry, error) 1193 1193 + // Shutdown the database client and all subscriptions. 1194 1194 + Shutdown() error 1175 1195 Subscribe(filter *Filter) *SubscribeResponse 1176 1196 // Subscribe with options. 1177 1197 SubscribeWithOpts(opts SubscribeOpts) *SubscribeResponse ··· 1218 1238 return res, err 1219 1239 } 1220 1240 1241 1241 + // Shutdown the database client and all subscriptions. 1242 1242 + func (_self *Db) Shutdown() error { 1243 1243 + _pointer := _self.ffiObject.incrementPointer("*Db") 1244 1244 + defer _self.ffiObject.decrementPointer() 1245 1245 + _, err := uniffiRustCallAsync[ShutdownError]( 1246 1246 + FfiConverterShutdownErrorINSTANCE, 1247 1247 + // completeFn 1248 1248 + func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 1249 1249 + C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 1250 1250 + return struct{}{} 1251 1251 + }, 1252 1252 + // liftFn 1253 1253 + func(_ struct{}) struct{} { return struct{}{} }, 1254 1254 + C.uniffi_iroh_streamplace_fn_method_db_shutdown( 1255 1255 + _pointer), 1256 1256 + // pollFn 1257 1257 + func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1258 1258 + C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 1259 1259 + }, 1260 1260 + // freeFn 1261 1261 + func(handle C.uint64_t) { 1262 1262 + C.ffi_iroh_streamplace_rust_future_free_void(handle) 1263 1263 + }, 1264 1264 + ) 1265 1265 + 1266 1266 + if err == nil { 1267 1267 + return nil 1268 1268 + } 1269 1269 + 1270 1270 + return err 1271 1271 + } 1272 1272 + 1221 1273 func (_self *Db) Subscribe(filter *Filter) *SubscribeResponse { 1222 1274 _pointer := _self.ffiObject.incrementPointer("*Db") 1223 1275 defer _self.ffiObject.decrementPointer() ··· 1591 1643 NodeScope() *WriteScope 1592 1644 // Send a segment to all subscribers of the given stream. 1593 1645 SendSegment(key string, data []byte) error 1646 1646 + // Shutdown the node, including the streaming system and the metadata db. 1647 1647 + Shutdown() error 1594 1648 // Subscribe to updates for a given stream from a remote node. 1595 1649 Subscribe(key string, remoteId *PublicKey) error 1596 1650 // Get this node's ticket. ··· 1830 1884 func(_ struct{}) struct{} { return struct{}{} }, 1831 1885 C.uniffi_iroh_streamplace_fn_method_node_send_segment( 1832 1886 _pointer, FfiConverterStringINSTANCE.Lower(key), FfiConverterBytesINSTANCE.Lower(data)), 1887 1887 + // pollFn 1888 1888 + func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1889 1889 + C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 1890 1890 + }, 1891 1891 + // freeFn 1892 1892 + func(handle C.uint64_t) { 1893 1893 + C.ffi_iroh_streamplace_rust_future_free_void(handle) 1894 1894 + }, 1895 1895 + ) 1896 1896 + 1897 1897 + if err == nil { 1898 1898 + return nil 1899 1899 + } 1900 1900 + 1901 1901 + return err 1902 1902 + } 1903 1903 + 1904 1904 + // Shutdown the node, including the streaming system and the metadata db. 1905 1905 + func (_self *Node) Shutdown() error { 1906 1906 + _pointer := _self.ffiObject.incrementPointer("*Node") 1907 1907 + defer _self.ffiObject.decrementPointer() 1908 1908 + _, err := uniffiRustCallAsync[ShutdownError]( 1909 1909 + FfiConverterShutdownErrorINSTANCE, 1910 1910 + // completeFn 1911 1911 + func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 1912 1912 + C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 1913 1913 + return struct{}{} 1914 1914 + }, 1915 1915 + // liftFn 1916 1916 + func(_ struct{}) struct{} { return struct{}{} }, 1917 1917 + C.uniffi_iroh_streamplace_fn_method_node_shutdown( 1918 1918 + _pointer), 1833 1919 // pollFn 1834 1920 func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1835 1921 C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) ··· 2833 2919 } 2834 2920 } 2835 2921 2836 2836 - // An Error. 2837 2837 - type Error struct { 2838 2838 - err error 2839 2839 - } 2840 2840 - 2841 2841 - // Convience method to turn *Error into error 2842 2842 - // Avoiding treating nil pointer as non nil error interface 2843 2843 - func (err *Error) AsError() error { 2844 2844 - if err == nil { 2845 2845 - return nil 2846 2846 - } else { 2847 2847 - return err 2848 2848 - } 2849 2849 - } 2850 2850 - 2851 2851 - func (err Error) Error() string { 2852 2852 - return fmt.Sprintf("Error: %s", err.err.Error()) 2853 2853 - } 2854 2854 - 2855 2855 - func (err Error) Unwrap() error { 2856 2856 - return err.err 2857 2857 - } 2858 2858 - 2859 2859 - // Err* are used for checking error type with `errors.Is` 2860 2860 - var ErrErrorIrohBind = fmt.Errorf("ErrorIrohBind") 2861 2861 - var ErrErrorIrohConnect = fmt.Errorf("ErrorIrohConnect") 2862 2862 - var ErrErrorMissingConnection = fmt.Errorf("ErrorMissingConnection") 2863 2863 - var ErrErrorIrpc = fmt.Errorf("ErrorIrpc") 2864 2864 - 2865 2865 - // Variant structs 2866 2866 - type ErrorIrohBind struct { 2867 2867 - message string 2868 2868 - } 2869 2869 - 2870 2870 - func NewErrorIrohBind() *Error { 2871 2871 - return &Error{err: &ErrorIrohBind{}} 2872 2872 - } 2873 2873 - 2874 2874 - func (e ErrorIrohBind) destroy() { 2875 2875 - } 2876 2876 - 2877 2877 - func (err ErrorIrohBind) Error() string { 2878 2878 - return fmt.Sprintf("IrohBind: %s", err.message) 2879 2879 - } 2880 2880 - 2881 2881 - func (self ErrorIrohBind) Is(target error) bool { 2882 2882 - return target == ErrErrorIrohBind 2883 2883 - } 2884 2884 - 2885 2885 - type ErrorIrohConnect struct { 2886 2886 - message string 2887 2887 - } 2888 2888 - 2889 2889 - func NewErrorIrohConnect() *Error { 2890 2890 - return &Error{err: &ErrorIrohConnect{}} 2891 2891 - } 2892 2892 - 2893 2893 - func (e ErrorIrohConnect) destroy() { 2894 2894 - } 2895 2895 - 2896 2896 - func (err ErrorIrohConnect) Error() string { 2897 2897 - return fmt.Sprintf("IrohConnect: %s", err.message) 2898 2898 - } 2899 2899 - 2900 2900 - func (self ErrorIrohConnect) Is(target error) bool { 2901 2901 - return target == ErrErrorIrohConnect 2902 2902 - } 2903 2903 - 2904 2904 - type ErrorMissingConnection struct { 2905 2905 - message string 2906 2906 - } 2907 2907 - 2908 2908 - func NewErrorMissingConnection() *Error { 2909 2909 - return &Error{err: &ErrorMissingConnection{}} 2910 2910 - } 2911 2911 - 2912 2912 - func (e ErrorMissingConnection) destroy() { 2913 2913 - } 2914 2914 - 2915 2915 - func (err ErrorMissingConnection) Error() string { 2916 2916 - return fmt.Sprintf("MissingConnection: %s", err.message) 2917 2917 - } 2918 2918 - 2919 2919 - func (self ErrorMissingConnection) Is(target error) bool { 2920 2920 - return target == ErrErrorMissingConnection 2921 2921 - } 2922 2922 - 2923 2923 - type ErrorIrpc struct { 2924 2924 - message string 2925 2925 - } 2926 2926 - 2927 2927 - func NewErrorIrpc() *Error { 2928 2928 - return &Error{err: &ErrorIrpc{}} 2929 2929 - } 2930 2930 - 2931 2931 - func (e ErrorIrpc) destroy() { 2932 2932 - } 2933 2933 - 2934 2934 - func (err ErrorIrpc) Error() string { 2935 2935 - return fmt.Sprintf("Irpc: %s", err.message) 2936 2936 - } 2937 2937 - 2938 2938 - func (self ErrorIrpc) Is(target error) bool { 2939 2939 - return target == ErrErrorIrpc 2940 2940 - } 2941 2941 - 2942 2942 - type FfiConverterError struct{} 2943 2943 - 2944 2944 - var FfiConverterErrorINSTANCE = FfiConverterError{} 2945 2945 - 2946 2946 - func (c FfiConverterError) Lift(eb RustBufferI) *Error { 2947 2947 - return LiftFromRustBuffer[*Error](c, eb) 2948 2948 - } 2949 2949 - 2950 2950 - func (c FfiConverterError) Lower(value *Error) C.RustBuffer { 2951 2951 - return LowerIntoRustBuffer[*Error](c, value) 2952 2952 - } 2953 2953 - 2954 2954 - func (c FfiConverterError) Read(reader io.Reader) *Error { 2955 2955 - errorID := readUint32(reader) 2956 2956 - 2957 2957 - message := FfiConverterStringINSTANCE.Read(reader) 2958 2958 - switch errorID { 2959 2959 - case 1: 2960 2960 - return &Error{&ErrorIrohBind{message}} 2961 2961 - case 2: 2962 2962 - return &Error{&ErrorIrohConnect{message}} 2963 2963 - case 3: 2964 2964 - return &Error{&ErrorMissingConnection{message}} 2965 2965 - case 4: 2966 2966 - return &Error{&ErrorIrpc{message}} 2967 2967 - default: 2968 2968 - panic(fmt.Sprintf("Unknown error code %d in FfiConverterError.Read()", errorID)) 2969 2969 - } 2970 2970 - 2971 2971 - } 2972 2972 - 2973 2973 - func (c FfiConverterError) Write(writer io.Writer, value *Error) { 2974 2974 - switch variantValue := value.err.(type) { 2975 2975 - case *ErrorIrohBind: 2976 2976 - writeInt32(writer, 1) 2977 2977 - case *ErrorIrohConnect: 2978 2978 - writeInt32(writer, 2) 2979 2979 - case *ErrorMissingConnection: 2980 2980 - writeInt32(writer, 3) 2981 2981 - case *ErrorIrpc: 2982 2982 - writeInt32(writer, 4) 2983 2983 - default: 2984 2984 - _ = variantValue 2985 2985 - panic(fmt.Sprintf("invalid error value `%v` in FfiConverterError.Write", value)) 2986 2986 - } 2987 2987 - } 2988 2988 - 2989 2989 - type FfiDestroyerError struct{} 2990 2990 - 2991 2991 - func (_ FfiDestroyerError) Destroy(value *Error) { 2992 2992 - switch variantValue := value.err.(type) { 2993 2993 - case ErrorIrohBind: 2994 2994 - variantValue.destroy() 2995 2995 - case ErrorIrohConnect: 2996 2996 - variantValue.destroy() 2997 2997 - case ErrorMissingConnection: 2998 2998 - variantValue.destroy() 2999 2999 - case ErrorIrpc: 3000 3000 - variantValue.destroy() 3001 3001 - default: 3002 3002 - _ = variantValue 3003 3003 - panic(fmt.Sprintf("invalid error value `%v` in FfiDestroyerError.Destroy", value)) 3004 3004 - } 3005 3005 - } 3006 3006 - 3007 2922 // Error joining peers. 3008 2923 type JoinPeersError struct { 3009 2924 err error ··· 3628 3543 default: 3629 3544 _ = variantValue 3630 3545 panic(fmt.Sprintf("invalid error value `%v` in FfiDestroyerSpError.Destroy", value)) 3546 3546 + } 3547 3547 + } 3548 3548 + 3549 3549 + // Error shutting down the database. 3550 3550 + // 3551 3551 + // This can occur if the db is already shut down or if there is an internal error. 3552 3552 + type ShutdownError struct { 3553 3553 + err error 3554 3554 + } 3555 3555 + 3556 3556 + // Convience method to turn *ShutdownError into error 3557 3557 + // Avoiding treating nil pointer as non nil error interface 3558 3558 + func (err *ShutdownError) AsError() error { 3559 3559 + if err == nil { 3560 3560 + return nil 3561 3561 + } else { 3562 3562 + return err 3563 3563 + } 3564 3564 + } 3565 3565 + 3566 3566 + func (err ShutdownError) Error() string { 3567 3567 + return fmt.Sprintf("ShutdownError: %s", err.err.Error()) 3568 3568 + } 3569 3569 + 3570 3570 + func (err ShutdownError) Unwrap() error { 3571 3571 + return err.err 3572 3572 + } 3573 3573 + 3574 3574 + // Err* are used for checking error type with `errors.Is` 3575 3575 + var ErrShutdownErrorIrpc = fmt.Errorf("ShutdownErrorIrpc") 3576 3576 + 3577 3577 + // Variant structs 3578 3578 + // Error during the shutdown operation. 3579 3579 + type ShutdownErrorIrpc struct { 3580 3580 + Message string 3581 3581 + } 3582 3582 + 3583 3583 + // Error during the shutdown operation. 3584 3584 + func NewShutdownErrorIrpc( 3585 3585 + message string, 3586 3586 + ) *ShutdownError { 3587 3587 + return &ShutdownError{err: &ShutdownErrorIrpc{ 3588 3588 + Message: message}} 3589 3589 + } 3590 3590 + 3591 3591 + func (e ShutdownErrorIrpc) destroy() { 3592 3592 + FfiDestroyerString{}.Destroy(e.Message) 3593 3593 + } 3594 3594 + 3595 3595 + func (err ShutdownErrorIrpc) Error() string { 3596 3596 + return fmt.Sprint("Irpc", 3597 3597 + ": ", 3598 3598 + 3599 3599 + "Message=", 3600 3600 + err.Message, 3601 3601 + ) 3602 3602 + } 3603 3603 + 3604 3604 + func (self ShutdownErrorIrpc) Is(target error) bool { 3605 3605 + return target == ErrShutdownErrorIrpc 3606 3606 + } 3607 3607 + 3608 3608 + type FfiConverterShutdownError struct{} 3609 3609 + 3610 3610 + var FfiConverterShutdownErrorINSTANCE = FfiConverterShutdownError{} 3611 3611 + 3612 3612 + func (c FfiConverterShutdownError) Lift(eb RustBufferI) *ShutdownError { 3613 3613 + return LiftFromRustBuffer[*ShutdownError](c, eb) 3614 3614 + } 3615 3615 + 3616 3616 + func (c FfiConverterShutdownError) Lower(value *ShutdownError) C.RustBuffer { 3617 3617 + return LowerIntoRustBuffer[*ShutdownError](c, value) 3618 3618 + } 3619 3619 + 3620 3620 + func (c FfiConverterShutdownError) Read(reader io.Reader) *ShutdownError { 3621 3621 + errorID := readUint32(reader) 3622 3622 + 3623 3623 + switch errorID { 3624 3624 + case 1: 3625 3625 + return &ShutdownError{&ShutdownErrorIrpc{ 3626 3626 + Message: FfiConverterStringINSTANCE.Read(reader), 3627 3627 + }} 3628 3628 + default: 3629 3629 + panic(fmt.Sprintf("Unknown error code %d in FfiConverterShutdownError.Read()", errorID)) 3630 3630 + } 3631 3631 + } 3632 3632 + 3633 3633 + func (c FfiConverterShutdownError) Write(writer io.Writer, value *ShutdownError) { 3634 3634 + switch variantValue := value.err.(type) { 3635 3635 + case *ShutdownErrorIrpc: 3636 3636 + writeInt32(writer, 1) 3637 3637 + FfiConverterStringINSTANCE.Write(writer, variantValue.Message) 3638 3638 + default: 3639 3639 + _ = variantValue 3640 3640 + panic(fmt.Sprintf("invalid error value `%v` in FfiConverterShutdownError.Write", value)) 3641 3641 + } 3642 3642 + } 3643 3643 + 3644 3644 + type FfiDestroyerShutdownError struct{} 3645 3645 + 3646 3646 + func (_ FfiDestroyerShutdownError) Destroy(value *ShutdownError) { 3647 3647 + switch variantValue := value.err.(type) { 3648 3648 + case ShutdownErrorIrpc: 3649 3649 + variantValue.destroy() 3650 3650 + default: 3651 3651 + _ = variantValue 3652 3652 + panic(fmt.Sprintf("invalid error value `%v` in FfiDestroyerShutdownError.Destroy", value)) 3631 3653 } 3632 3654 } 3633 3655

+22

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h

··· 457 457 uint64_t uniffi_iroh_streamplace_fn_method_db_iter_with_opts(void* ptr, void* filter 458 458 ); 459 459 #endif 460 460 + #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_SHUTDOWN 461 461 + #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_SHUTDOWN 462 462 + uint64_t uniffi_iroh_streamplace_fn_method_db_shutdown(void* ptr 463 463 + ); 464 464 + #endif 460 465 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_SUBSCRIBE 461 466 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_SUBSCRIBE 462 467 void* uniffi_iroh_streamplace_fn_method_db_subscribe(void* ptr, void* filter, RustCallStatus *out_status ··· 596 601 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_NODE_SEND_SEGMENT 597 602 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_NODE_SEND_SEGMENT 598 603 uint64_t uniffi_iroh_streamplace_fn_method_node_send_segment(void* ptr, RustBuffer key, RustBuffer data 604 604 + ); 605 605 + #endif 606 606 + #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_NODE_SHUTDOWN 607 607 + #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_NODE_SHUTDOWN 608 608 + uint64_t uniffi_iroh_streamplace_fn_method_node_shutdown(void* ptr 599 609 ); 600 610 #endif 601 611 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_NODE_SUBSCRIBE ··· 1048 1058 1049 1059 ); 1050 1060 #endif 1061 1061 + #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_SHUTDOWN 1062 1062 + #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_SHUTDOWN 1063 1063 + uint16_t uniffi_iroh_streamplace_checksum_method_db_shutdown(void 1064 1064 + 1065 1065 + ); 1066 1066 + #endif 1051 1067 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_SUBSCRIBE 1052 1068 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_SUBSCRIBE 1053 1069 uint16_t uniffi_iroh_streamplace_checksum_method_db_subscribe(void ··· 1147 1163 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_NODE_SEND_SEGMENT 1148 1164 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_NODE_SEND_SEGMENT 1149 1165 uint16_t uniffi_iroh_streamplace_checksum_method_node_send_segment(void 1166 1166 + 1167 1167 + ); 1168 1168 + #endif 1169 1169 + #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_NODE_SHUTDOWN 1170 1170 + #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_NODE_SHUTDOWN 1171 1171 + uint16_t uniffi_iroh_streamplace_checksum_method_node_shutdown(void 1150 1172 1151 1173 ); 1152 1174 #endif

+3 -2

pkg/iroh/iroh_test/main.go

··· 71 71 fmt.Printf("Iter items: %+v\n", items3) 72 72 73 73 go func() { 74 74 - time.Sleep(5 * time.Second) 75 75 - node.Shutdown() // or whatever your shutdown method is 74 74 + time.Sleep(5 * time.Second) 75 75 + err := node.Shutdown() 76 76 + panicIfErr(err) 76 77 }() 77 78 78 79 sub := db.Subscribe(iroh.NewFilter())

-19

rust/iroh-streamplace/src/error.rs

··· 1 1 - /// An Error. 2 2 - #[derive(Debug, snafu::Snafu, uniffi::Error)] 3 3 - #[uniffi(flat_error)] 4 4 - #[snafu(visibility(pub(crate)))] 5 5 - pub enum Error { 6 6 - #[snafu(display("Bind failure"), context(false))] 7 7 - IrohBind { 8 8 - source: iroh::endpoint::BindError, 9 9 - }, 10 10 - #[snafu(display("Failed to connect"), context(false))] 11 11 - IrohConnect { 12 12 - source: iroh::endpoint::ConnectError, 13 13 - }, 14 14 - MissingConnection, 15 15 - #[snafu(display("RPC error"), context(false))] 16 16 - Irpc { 17 17 - source: irpc::Error, 18 18 - }, 19 19 - }

+802 -2

rust/iroh-streamplace/src/lib.rs

··· 1 1 uniffi::setup_scaffolding!(); 2 2 3 3 pub mod c2pa; 4 4 - pub mod error; 5 5 - pub mod node; 6 4 pub mod node_addr; 7 5 pub mod public_key; 6 6 + 7 7 + use std::sync::LazyLock; 8 8 + 9 9 + mod db; 10 10 + pub use db::*; 11 11 + #[cfg(test)] 12 12 + mod tests; 13 13 + 14 14 + /// Lazily initialized Tokio runtime for use in uniffi methods that need a runtime. 15 15 + static RUNTIME: LazyLock<tokio::runtime::Runtime> = 16 16 + LazyLock::new(|| tokio::runtime::Runtime::new().unwrap()); 17 17 + 18 18 + use std::{ 19 19 + collections::{BTreeMap, BTreeSet, HashSet}, 20 20 + str::FromStr, 21 21 + sync::Arc, 22 22 + }; 23 23 + 24 24 + use bytes::Bytes; 25 25 + use iroh::{NodeId, PublicKey, RelayMode, SecretKey, Watcher}; 26 26 + use iroh_base::ticket::NodeTicket; 27 27 + use iroh_gossip::{net::Gossip, proto::TopicId}; 28 28 + use irpc::{WithChannels, rpc::RemoteService}; 29 29 + use irpc_iroh::{IrohProtocol, IrohRemoteConnection}; 30 30 + use n0_future::future::Boxed; 31 31 + 32 32 + mod rpc { 33 33 + //! Protocol API 34 34 + use bytes::Bytes; 35 35 + use iroh::NodeId; 36 36 + use irpc::{channel::oneshot, rpc_requests}; 37 37 + use serde::{Deserialize, Serialize}; 38 38 + 39 39 + pub const ALPN: &[u8] = b"/iroh/streamplace/1"; 40 40 + 41 41 + /// Subscribe to the given `key` 42 42 + #[derive(Debug, Serialize, Deserialize)] 43 43 + pub struct Subscribe { 44 44 + pub key: String, 45 45 + // TODO: verify 46 46 + pub remote_id: NodeId, 47 47 + } 48 48 + 49 49 + /// Unsubscribe from the given `key` 50 50 + #[derive(Debug, Serialize, Deserialize)] 51 51 + pub struct Unsubscribe { 52 52 + pub key: String, 53 53 + // TODO: verify 54 54 + pub remote_id: NodeId, 55 55 + } 56 56 + 57 57 + // #[derive(Debug, Serialize, Deserialize)] 58 58 + // pub struct SendSegment { 59 59 + // pub key: String, 60 60 + // pub data: Bytes, 61 61 + // } 62 62 + 63 63 + #[derive(Debug, Clone, Serialize, Deserialize)] 64 64 + pub struct RecvSegment { 65 65 + pub key: String, 66 66 + pub data: Bytes, 67 67 + } 68 68 + 69 69 + // Use the macro to generate both the Protocol and Message enums 70 70 + // plus implement Channels for each type 71 71 + #[rpc_requests(message = Message)] 72 72 + #[derive(Serialize, Deserialize, Debug)] 73 73 + pub enum Protocol { 74 74 + #[rpc(tx=oneshot::Sender<()>)] 75 75 + Subscribe(Subscribe), 76 76 + #[rpc(tx=oneshot::Sender<()>)] 77 77 + Unsubscribe(Unsubscribe), 78 78 + #[rpc(tx=oneshot::Sender<()>)] 79 79 + RecvSegment(RecvSegment), 80 80 + } 81 81 + } 82 82 + 83 83 + mod api { 84 84 + //! Protocol API 85 85 + use bytes::Bytes; 86 86 + use iroh::{NodeAddr, NodeId}; 87 87 + use irpc::{channel::oneshot, rpc_requests}; 88 88 + use serde::{Deserialize, Serialize}; 89 89 + 90 90 + /// Subscribe to the given `key` 91 91 + #[derive(Debug, Serialize, Deserialize)] 92 92 + pub struct Subscribe { 93 93 + pub key: String, 94 94 + // TODO: verify 95 95 + pub remote_id: NodeId, 96 96 + } 97 97 + 98 98 + /// Unsubscribe from the given `key` 99 99 + #[derive(Debug, Serialize, Deserialize)] 100 100 + pub struct Unsubscribe { 101 101 + pub key: String, 102 102 + // TODO: verify 103 103 + pub remote_id: NodeId, 104 104 + } 105 105 + 106 106 + #[derive(Debug, Serialize, Deserialize)] 107 107 + pub struct SendSegment { 108 108 + pub key: String, 109 109 + pub data: Bytes, 110 110 + } 111 111 + 112 112 + #[derive(Debug, Serialize, Deserialize)] 113 113 + pub struct JoinPeers { 114 114 + pub peers: Vec<NodeAddr>, 115 115 + } 116 116 + 117 117 + #[derive(Debug, Serialize, Deserialize)] 118 118 + pub struct AddTickets { 119 119 + pub peers: Vec<NodeAddr>, 120 120 + } 121 121 + 122 122 + #[derive(Debug, Serialize, Deserialize)] 123 123 + pub struct GetNodeAddr; 124 124 + 125 125 + #[derive(Debug, Serialize, Deserialize)] 126 126 + pub struct Shutdown; 127 127 + 128 128 + // Use the macro to generate both the Protocol and Message enums 129 129 + // plus implement Channels for each type 130 130 + #[rpc_requests(message = Message)] 131 131 + #[derive(Serialize, Deserialize, Debug)] 132 132 + pub enum Protocol { 133 133 + #[rpc(tx=oneshot::Sender<()>)] 134 134 + Subscribe(Subscribe), 135 135 + #[rpc(tx=oneshot::Sender<()>)] 136 136 + Unsubscribe(Unsubscribe), 137 137 + #[rpc(tx=oneshot::Sender<()>)] 138 138 + SendSegment(SendSegment), 139 139 + #[rpc(tx=oneshot::Sender<()>)] 140 140 + JoinPeers(JoinPeers), 141 141 + #[rpc(tx=oneshot::Sender<()>)] 142 142 + AddTickets(AddTickets), 143 143 + #[rpc(tx=oneshot::Sender<NodeAddr>)] 144 144 + GetNodeAddr(GetNodeAddr), 145 145 + #[rpc(tx=oneshot::Sender<()>)] 146 146 + Shutdown(Shutdown), 147 147 + } 148 148 + } 149 149 + use api::{Message as ApiMessage, Protocol as ApiProtocol}; 150 150 + use n0_future::{FuturesUnordered, StreamExt}; 151 151 + use rpc::{Message as RpcMessage, Protocol as RpcProtocol}; 152 152 + use snafu::Snafu; 153 153 + use tracing::{Instrument, debug, error, trace, trace_span, warn}; 154 154 + 155 155 + use crate::rpc::RecvSegment; 156 156 + 157 157 + pub(crate) enum HandlerMode { 158 158 + Sender, 159 159 + Forwarder, 160 160 + Receiver(Arc<dyn DataHandler>), 161 161 + } 162 162 + 163 163 + impl HandlerMode { 164 164 + pub fn mode_str(&self) -> &'static str { 165 165 + match self { 166 166 + HandlerMode::Sender => "sender", 167 167 + HandlerMode::Forwarder => "forwarder", 168 168 + HandlerMode::Receiver(_) => "receiver", 169 169 + } 170 170 + } 171 171 + } 172 172 + 173 173 + type Tasks = FuturesUnordered<Boxed<(NodeId, Result<(), RpcTaskError>)>>; 174 174 + 175 175 + /// Actor that contains both a kv db for metadata and a handler for the rpc protocol. 176 176 + /// 177 177 + /// This can be used both for sender and receiver nodes. Sender nodes will just set the 178 178 + /// handler to None. 179 179 + struct Actor { 180 180 + /// Receiver for rpc messages from remote nodes 181 181 + rpc_rx: tokio::sync::mpsc::Receiver<RpcMessage>, 182 182 + /// Receiver for API messages from the user 183 183 + api_rx: tokio::sync::mpsc::Receiver<ApiMessage>, 184 184 + /// nodes I need to send to for each stream 185 185 + subscribers: BTreeMap<String, BTreeSet<NodeId>>, 186 186 + /// nodes I am subscribed to 187 187 + subscriptions: BTreeMap<String, NodeId>, 188 188 + /// lightweight typed connection pool 189 189 + connections: ConnectionPool, 190 190 + /// How to handle incoming data 191 191 + handler: HandlerMode, 192 192 + /// Iroh protocol router, I need to keep it around to keep the protocol alive 193 193 + router: iroh::protocol::Router, 194 194 + /// Metadata db 195 195 + client: db::Db, 196 196 + /// Write scope for this node for the metadata db 197 197 + write: db::WriteScope, 198 198 + /// Ongoing tasks 199 199 + tasks: Tasks, 200 200 + /// Configuration, needed for timeouts etc. 201 201 + config: Arc<Config>, 202 202 + } 203 203 + 204 204 + #[derive(Debug, Clone)] 205 205 + struct Connection { 206 206 + id: NodeId, 207 207 + rpc: irpc::Client<RpcProtocol>, 208 208 + } 209 209 + 210 210 + #[derive(Debug, Snafu)] 211 211 + enum RpcTaskError { 212 212 + #[snafu(transparent)] 213 213 + Task { source: irpc::Error }, 214 214 + #[snafu(transparent)] 215 215 + Timeout { source: tokio::time::error::Elapsed }, 216 216 + } 217 217 + 218 218 + struct ConnectionPool { 219 219 + endpoint: iroh::Endpoint, 220 220 + connections: BTreeMap<NodeId, Connection>, 221 221 + } 222 222 + 223 223 + impl ConnectionPool { 224 224 + fn new(endpoint: iroh::Endpoint) -> Self { 225 225 + Self { 226 226 + endpoint, 227 227 + connections: BTreeMap::new(), 228 228 + } 229 229 + } 230 230 + 231 231 + /// Cheap conn pool hack 232 232 + fn get(&mut self, remote: &NodeId) -> Connection { 233 233 + if !self.connections.contains_key(remote) { 234 234 + let conn = IrohRemoteConnection::new( 235 235 + self.endpoint.clone(), 236 236 + (*remote).into(), 237 237 + rpc::ALPN.to_vec(), 238 238 + ); 239 239 + let conn = Connection { 240 240 + rpc: irpc::Client::boxed(conn), 241 241 + id: *remote, 242 242 + }; 243 243 + self.connections.insert(*remote, conn); 244 244 + } 245 245 + self.connections 246 246 + .get_mut(remote) 247 247 + .expect("just inserted") 248 248 + .clone() 249 249 + } 250 250 + 251 251 + fn remove(&mut self, remote: &NodeId) { 252 252 + self.connections.remove(remote); 253 253 + } 254 254 + } 255 255 + 256 256 + impl Actor { 257 257 + pub async fn spawn( 258 258 + endpoint: iroh::Endpoint, 259 259 + topic: iroh_gossip::proto::TopicId, 260 260 + config: Config, 261 261 + handler: HandlerMode, 262 262 + ) -> Result<(Node, impl Future<Output = ()>), iroh_gossip::api::ApiError> { 263 263 + let (rpc_tx, rpc_rx) = tokio::sync::mpsc::channel::<RpcMessage>(32); 264 264 + let (api_tx, api_rx) = tokio::sync::mpsc::channel::<ApiMessage>(32); 265 265 + let gossip = Gossip::builder().spawn(endpoint.clone()); 266 266 + let id = endpoint.node_id(); 267 267 + let router = iroh::protocol::Router::builder(endpoint.clone()) 268 268 + .accept(iroh_gossip::ALPN, gossip.clone()) 269 269 + .accept( 270 270 + rpc::ALPN, 271 271 + IrohProtocol::new(rpc::Protocol::remote_handler(rpc_tx.into())), 272 272 + ) 273 273 + .spawn(); 274 274 + let topic = gossip.subscribe(topic, vec![]).await?; 275 275 + let secret = router.endpoint().secret_key().clone(); 276 276 + let db_config = Default::default(); 277 277 + let client = iroh_smol_kv::Client::local(topic, db_config); 278 278 + let write = db::WriteScope::new(client.write(secret.clone())); 279 279 + let client = db::Db::new(client); 280 280 + let actor = Self { 281 281 + rpc_rx, 282 282 + api_rx, 283 283 + subscribers: BTreeMap::new(), 284 284 + subscriptions: BTreeMap::new(), 285 285 + connections: ConnectionPool::new(router.endpoint().clone()), 286 286 + handler, 287 287 + router, 288 288 + write: write.clone(), 289 289 + client: client.clone(), 290 290 + tasks: FuturesUnordered::new(), 291 291 + config: Arc::new(config), 292 292 + }; 293 293 + let api = Node { 294 294 + client: Arc::new(client), 295 295 + write: Arc::new(write), 296 296 + api: irpc::Client::local(api_tx), 297 297 + }; 298 298 + Ok(( 299 299 + api, 300 300 + actor 301 301 + .run() 302 302 + .instrument(trace_span!("actor", id=%id.fmt_short())), 303 303 + )) 304 304 + } 305 305 + 306 306 + async fn run(mut self) { 307 307 + loop { 308 308 + tokio::select! { 309 309 + msg = self.rpc_rx.recv() => { 310 310 + let Some(msg) = msg else { 311 311 + error!("rpc channel closed"); 312 312 + break; 313 313 + }; 314 314 + self.handle_rpc(msg).instrument(trace_span!("rpc")).await; 315 315 + } 316 316 + msg = self.api_rx.recv() => { 317 317 + let Some(msg) = msg else { 318 318 + break; 319 319 + }; 320 320 + if let Some(shutdown) = self.handle_api(msg).instrument(trace_span!("api")).await { 321 321 + shutdown.send(()).await.ok(); 322 322 + break; 323 323 + } 324 324 + } 325 325 + res = self.tasks.next(), if !self.tasks.is_empty() => { 326 326 + let Some((remote_id, res)) = res else { 327 327 + error!("task finished but no result"); 328 328 + break; 329 329 + }; 330 330 + match res { 331 331 + Ok(()) => {} 332 332 + Err(RpcTaskError::Timeout { source }) => { 333 333 + warn!("call to {remote_id} timed out: {source}"); 334 334 + } 335 335 + Err(RpcTaskError::Task { source }) => { 336 336 + warn!("call to {remote_id} failed: {source}"); 337 337 + } 338 338 + } 339 339 + self.connections.remove(&remote_id); 340 340 + } 341 341 + } 342 342 + } 343 343 + } 344 344 + 345 345 + async fn update_subscriber_meta(&mut self, key: &str) { 346 346 + let n = self 347 347 + .subscribers 348 348 + .get(key) 349 349 + .map(|s| s.len()) 350 350 + .unwrap_or_default(); 351 351 + let v = n.to_string().into_bytes(); 352 352 + self.write 353 353 + .put_impl(Some(key.as_bytes().to_vec()), b"subscribers", v.into()) 354 354 + .await 355 355 + .ok(); 356 356 + } 357 357 + 358 358 + /// Requests from remote nodes 359 359 + async fn handle_rpc(&mut self, msg: RpcMessage) { 360 360 + match msg { 361 361 + RpcMessage::Subscribe(msg) => { 362 362 + trace!("{:?}", msg.inner); 363 363 + let WithChannels { 364 364 + tx, 365 365 + inner: rpc::Subscribe { key, remote_id }, 366 366 + .. 367 367 + } = msg; 368 368 + self.subscribers 369 369 + .entry(key.clone()) 370 370 + .or_default() 371 371 + .insert(remote_id); 372 372 + self.update_subscriber_meta(&key).await; 373 373 + tx.send(()).await.ok(); 374 374 + } 375 375 + RpcMessage::Unsubscribe(msg) => { 376 376 + debug!("{:?}", msg.inner); 377 377 + let WithChannels { 378 378 + tx, 379 379 + inner: rpc::Unsubscribe { key, remote_id }, 380 380 + .. 381 381 + } = msg; 382 382 + if let Some(e) = self.subscribers.get_mut(&key) 383 383 + && !e.remove(&remote_id) 384 384 + { 385 385 + warn!( 386 386 + "unsubscribe: no subscription for {} from {}", 387 387 + key, remote_id 388 388 + ); 389 389 + } 390 390 + if let Some(subscriptions) = self.subscribers.get(&key) 391 391 + && subscriptions.is_empty() 392 392 + { 393 393 + self.subscribers.remove(&key); 394 394 + } 395 395 + self.update_subscriber_meta(&key).await; 396 396 + tx.send(()).await.ok(); 397 397 + } 398 398 + RpcMessage::RecvSegment(msg) => { 399 399 + trace!("{:?}", msg.inner); 400 400 + let WithChannels { 401 401 + tx, 402 402 + inner: rpc::RecvSegment { key, data }, 403 403 + .. 404 404 + } = msg; 405 405 + match &self.handler { 406 406 + HandlerMode::Sender => { 407 407 + warn!("received segment but in sender mode"); 408 408 + } 409 409 + HandlerMode::Forwarder => { 410 410 + if let Some(remotes) = self.subscribers.get(&key) { 411 411 + Self::handle_send( 412 412 + &mut self.tasks, 413 413 + &mut self.connections, 414 414 + &self.config, 415 415 + key, 416 416 + data, 417 417 + remotes, 418 418 + ); 419 419 + } else { 420 420 + trace!("no subscribers for stream {}", key); 421 421 + } 422 422 + } 423 423 + HandlerMode::Receiver(handler) => { 424 424 + if self.subscriptions.contains_key(&key) { 425 425 + handler.handle_data(key, data.to_vec()).await; 426 426 + } else { 427 427 + warn!("received segment for unsubscribed key: {}", key); 428 428 + } 429 429 + } 430 430 + }; 431 431 + tx.send(()).await.ok(); 432 432 + } 433 433 + } 434 434 + } 435 435 + 436 436 + async fn handle_api(&mut self, msg: ApiMessage) -> Option<irpc::channel::oneshot::Sender<()>> { 437 437 + match msg { 438 438 + ApiMessage::SendSegment(msg) => { 439 439 + trace!("{:?}", msg.inner); 440 440 + let WithChannels { 441 441 + tx, 442 442 + inner: api::SendSegment { key, data }, 443 443 + .. 444 444 + } = msg; 445 445 + if let Some(remotes) = self.subscribers.get(&key) { 446 446 + Self::handle_send( 447 447 + &mut self.tasks, 448 448 + &mut self.connections, 449 449 + &self.config, 450 450 + key, 451 451 + data, 452 452 + remotes, 453 453 + ); 454 454 + } else { 455 455 + trace!("no subscribers for stream {}", key); 456 456 + } 457 457 + tx.send(()).await.ok(); 458 458 + } 459 459 + ApiMessage::Subscribe(msg) => { 460 460 + trace!("{:?}", msg.inner); 461 461 + let WithChannels { 462 462 + tx, 463 463 + inner: api::Subscribe { key, remote_id }, 464 464 + .. 465 465 + } = msg; 466 466 + let conn = self.connections.get(&remote_id); 467 467 + conn.rpc 468 468 + .rpc(rpc::Subscribe { 469 469 + key: key.clone(), 470 470 + remote_id: self.node_id(), 471 471 + }) 472 472 + .await 473 473 + .ok(); 474 474 + self.subscriptions.insert(key, remote_id); 475 475 + tx.send(()).await.ok(); 476 476 + } 477 477 + ApiMessage::Unsubscribe(msg) => { 478 478 + trace!("{:?}", msg.inner); 479 479 + let WithChannels { 480 480 + tx, 481 481 + inner: api::Unsubscribe { key, remote_id }, 482 482 + .. 483 483 + } = msg; 484 484 + let conn = self.connections.get(&remote_id); 485 485 + conn.rpc 486 486 + .rpc(rpc::Unsubscribe { 487 487 + key: key.clone(), 488 488 + remote_id: self.node_id(), 489 489 + }) 490 490 + .await 491 491 + .ok(); 492 492 + self.subscriptions.remove(&key); 493 493 + tx.send(()).await.ok(); 494 494 + } 495 495 + ApiMessage::AddTickets(msg) => { 496 496 + trace!("{:?}", msg.inner); 497 497 + let WithChannels { 498 498 + tx, 499 499 + inner: api::AddTickets { peers }, 500 500 + .. 501 501 + } = msg; 502 502 + for addr in &peers { 503 503 + self.router.endpoint().add_node_addr(addr.clone()).ok(); 504 504 + } 505 505 + // self.client.inner().join_peers(ids).await.ok(); 506 506 + tx.send(()).await.ok(); 507 507 + } 508 508 + ApiMessage::JoinPeers(msg) => { 509 509 + trace!("{:?}", msg.inner); 510 510 + let WithChannels { 511 511 + tx, 512 512 + inner: api::JoinPeers { peers }, 513 513 + .. 514 514 + } = msg; 515 515 + let ids = peers 516 516 + .iter() 517 517 + .map(|a| a.node_id) 518 518 + .filter(|id| *id != self.node_id()) 519 519 + .collect::<HashSet<_>>(); 520 520 + for addr in &peers { 521 521 + self.router.endpoint().add_node_addr(addr.clone()).ok(); 522 522 + } 523 523 + self.client.inner().join_peers(ids).await.ok(); 524 524 + tx.send(()).await.ok(); 525 525 + } 526 526 + ApiMessage::GetNodeAddr(msg) => { 527 527 + trace!("{:?}", msg.inner); 528 528 + let WithChannels { tx, .. } = msg; 529 529 + if !self.config.disable_relay { 530 530 + // don't await home relay if we have disabled relays, this will hang forever 531 531 + self.router.endpoint().home_relay().initialized().await; 532 532 + } 533 533 + let addr = self.router.endpoint().node_addr().initialized().await; 534 534 + tx.send(addr).await.ok(); 535 535 + } 536 536 + ApiMessage::Shutdown(msg) => { 537 537 + return Some(msg.tx); 538 538 + } 539 539 + } 540 540 + None 541 541 + } 542 542 + 543 543 + fn handle_send( 544 544 + tasks: &mut Tasks, 545 545 + connections: &mut ConnectionPool, 546 546 + config: &Arc<Config>, 547 547 + key: String, 548 548 + data: Bytes, 549 549 + remotes: &BTreeSet<NodeId>, 550 550 + ) { 551 551 + let msg = rpc::RecvSegment { key, data }; 552 552 + for remote in remotes { 553 553 + trace!("sending to stream {}: {}", msg.key, remote); 554 554 + let conn = connections.get(remote); 555 555 + tasks.push(Box::pin(Self::forward_task( 556 556 + config.clone(), 557 557 + conn, 558 558 + msg.clone(), 559 559 + ))); 560 560 + } 561 561 + } 562 562 + 563 563 + async fn forward_task( 564 564 + config: Arc<Config>, 565 565 + conn: Connection, 566 566 + msg: RecvSegment, 567 567 + ) -> (NodeId, Result<(), RpcTaskError>) { 568 568 + let id = conn.id; 569 569 + let res = async move { 570 570 + tokio::time::timeout(config.max_send_duration, conn.rpc.rpc(msg)).await??; 571 571 + Ok(()) 572 572 + } 573 573 + .await; 574 574 + (id, res) 575 575 + } 576 576 + 577 577 + fn node_id(&self) -> PublicKey { 578 578 + self.router.endpoint().node_id() 579 579 + } 580 580 + } 581 581 + 582 582 + /// Iroh-streamplace node that can send, forward or receive stream segments. 583 583 + #[derive(Clone, uniffi::Object)] 584 584 + pub struct Node { 585 585 + client: Arc<db::Db>, 586 586 + write: Arc<db::WriteScope>, 587 587 + api: irpc::Client<ApiProtocol>, 588 588 + } 589 589 + 590 590 + impl Node { 591 591 + pub(crate) async fn new_in_runtime( 592 592 + config: Config, 593 593 + handler: HandlerMode, 594 594 + ) -> Result<Arc<Self>, CreateError> { 595 595 + let mode_str = Bytes::from(handler.mode_str()); 596 596 + let secret_key = 597 597 + SecretKey::from_bytes(&<[u8; 32]>::try_from(config.key.clone()).map_err(|e| { 598 598 + CreateError::PrivateKey { 599 599 + size: e.len() as u64, 600 600 + } 601 601 + })?); 602 602 + let topic = 603 603 + TopicId::from_bytes(<[u8; 32]>::try_from(config.topic.clone()).map_err(|e| { 604 604 + CreateError::Topic { 605 605 + size: e.len() as u64, 606 606 + } 607 607 + })?); 608 608 + let relay_mode = if config.disable_relay { 609 609 + RelayMode::Disabled 610 610 + } else { 611 611 + RelayMode::Default 612 612 + }; 613 613 + let endpoint = iroh::Endpoint::builder() 614 614 + .secret_key(secret_key) 615 615 + .relay_mode(relay_mode) 616 616 + .bind() 617 617 + .await 618 618 + .map_err(|e| CreateError::Bind { 619 619 + message: e.to_string(), 620 620 + })?; 621 621 + let (api, actor) = Actor::spawn(endpoint, topic, config, handler) 622 622 + .await 623 623 + .map_err(|e| CreateError::Subscribe { 624 624 + message: e.to_string(), 625 625 + })?; 626 626 + api.node_scope() 627 627 + .put_impl(Option::<Vec<u8>>::None, b"mode", mode_str) 628 628 + .await 629 629 + .ok(); 630 630 + tokio::spawn(actor); 631 631 + Ok(Arc::new(api)) 632 632 + } 633 633 + } 634 634 + 635 635 + /// DataHandler trait that is exported to go for receiving data callbacks. 636 636 + #[uniffi::export(with_foreign)] 637 637 + #[async_trait::async_trait] 638 638 + pub trait DataHandler: Send + Sync { 639 639 + async fn handle_data(&self, topic: String, data: Vec<u8>); 640 640 + } 641 641 + 642 642 + #[uniffi::export] 643 643 + impl Node { 644 644 + /// Create a new streamplace client node. 645 645 + #[uniffi::constructor] 646 646 + pub async fn sender(config: Config) -> Result<Arc<Self>, CreateError> { 647 647 + RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Sender)) 648 648 + } 649 649 + 650 650 + #[uniffi::constructor] 651 651 + pub async fn forwarder(config: Config) -> Result<Arc<Self>, CreateError> { 652 652 + RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Forwarder)) 653 653 + } 654 654 + 655 655 + #[uniffi::constructor] 656 656 + pub async fn receiver( 657 657 + config: Config, 658 658 + handler: Arc<dyn DataHandler>, 659 659 + ) -> Result<Arc<Self>, CreateError> { 660 660 + RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Receiver(handler))) 661 661 + } 662 662 + 663 663 + /// Get a handle to the db to watch for changes locally or globally. 664 664 + pub fn db(&self) -> Arc<db::Db> { 665 665 + self.client.clone() 666 666 + } 667 667 + 668 668 + /// Get a handle to the write scope for this node. 669 669 + /// 670 670 + /// This is equivalent to calling `db.write(...)` with the secret key used to create the node. 671 671 + pub fn node_scope(&self) -> Arc<db::WriteScope> { 672 672 + self.write.clone() 673 673 + } 674 674 + 675 675 + /// Subscribe to updates for a given stream from a remote node. 676 676 + pub async fn subscribe( 677 677 + &self, 678 678 + key: String, 679 679 + remote_id: Arc<crate::public_key::PublicKey>, 680 680 + ) -> Result<(), PutError> { 681 681 + self.api 682 682 + .rpc(api::Subscribe { 683 683 + key, 684 684 + remote_id: remote_id.as_ref().into(), 685 685 + }) 686 686 + .await 687 687 + .map_err(|e| PutError::Irpc { 688 688 + message: e.to_string(), 689 689 + }) 690 690 + } 691 691 + 692 692 + /// Unsubscribe from updates for a given stream from a remote node. 693 693 + pub async fn unsubscribe( 694 694 + &self, 695 695 + key: String, 696 696 + remote_id: Arc<crate::public_key::PublicKey>, 697 697 + ) -> Result<(), PutError> { 698 698 + self.api 699 699 + .rpc(api::Unsubscribe { 700 700 + key, 701 701 + remote_id: remote_id.as_ref().into(), 702 702 + }) 703 703 + .await 704 704 + .map_err(|e| PutError::Irpc { 705 705 + message: e.to_string(), 706 706 + }) 707 707 + } 708 708 + 709 709 + /// Send a segment to all subscribers of the given stream. 710 710 + pub async fn send_segment(&self, key: String, data: Vec<u8>) -> Result<(), PutError> { 711 711 + self.api 712 712 + .rpc(api::SendSegment { 713 713 + key, 714 714 + data: data.into(), 715 715 + }) 716 716 + .await 717 717 + .map_err(|e| PutError::Irpc { 718 718 + message: e.to_string(), 719 719 + }) 720 720 + } 721 721 + 722 722 + /// Join peers by their node tickets. 723 723 + pub async fn join_peers(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { 724 724 + let peers = peers 725 725 + .iter() 726 726 + .map(|p| NodeTicket::from_str(p)) 727 727 + .collect::<Result<Vec<_>, _>>() 728 728 + .map_err(|e| JoinPeersError::Ticket { 729 729 + message: e.to_string(), 730 730 + })?; 731 731 + let addrs = peers 732 732 + .iter() 733 733 + .map(|t| t.node_addr().clone()) 734 734 + .collect::<Vec<_>>(); 735 735 + self.api 736 736 + .rpc(api::JoinPeers { peers: addrs }) 737 737 + .await 738 738 + .map_err(|e| JoinPeersError::Irpc { 739 739 + message: e.to_string(), 740 740 + }) 741 741 + } 742 742 + 743 743 + /// Add tickets for remote peers 744 744 + pub async fn add_tickets(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { 745 745 + let peers = peers 746 746 + .iter() 747 747 + .map(|p| NodeTicket::from_str(p)) 748 748 + .collect::<Result<Vec<_>, _>>() 749 749 + .map_err(|e| JoinPeersError::Ticket { 750 750 + message: e.to_string(), 751 751 + })?; 752 752 + let addrs = peers 753 753 + .iter() 754 754 + .map(|t| t.node_addr().clone()) 755 755 + .collect::<Vec<_>>(); 756 756 + self.api 757 757 + .rpc(api::AddTickets { peers: addrs }) 758 758 + .await 759 759 + .map_err(|e| JoinPeersError::Irpc { 760 760 + message: e.to_string(), 761 761 + }) 762 762 + } 763 763 + 764 764 + /// Get this node's ticket. 765 765 + pub async fn ticket(&self) -> Result<String, PutError> { 766 766 + let addr = self 767 767 + .api 768 768 + .rpc(api::GetNodeAddr) 769 769 + .await 770 770 + .map_err(|e| PutError::Irpc { 771 771 + message: e.to_string(), 772 772 + })?; 773 773 + Ok(NodeTicket::from(addr).to_string()) 774 774 + } 775 775 + 776 776 + /// Get this node's node ID. 777 777 + pub async fn node_id(&self) -> Result<Arc<crate::public_key::PublicKey>, PutError> { 778 778 + let addr = self 779 779 + .api 780 780 + .rpc(api::GetNodeAddr) 781 781 + .await 782 782 + .map_err(|e| PutError::Irpc { 783 783 + message: e.to_string(), 784 784 + })?; 785 785 + Ok(Arc::new(addr.node_id.into())) 786 786 + } 787 787 + 788 788 + /// Shutdown the node, including the streaming system and the metadata db. 789 789 + pub async fn shutdown(&self) -> Result<(), ShutdownError> { 790 790 + // shut down both the streams and the db concurrently, even if one fails 791 791 + let (res1, res2) = tokio::join!(self.shutdown_streams(), self.client.shutdown()); 792 792 + res1?; 793 793 + res2?; 794 794 + Ok(()) 795 795 + } 796 796 + } 797 797 + 798 798 + impl Node { 799 799 + async fn shutdown_streams(&self) -> std::result::Result<(), ShutdownError> { 800 800 + self.api 801 801 + .rpc(api::Shutdown) 802 802 + .await 803 803 + .map_err(|e| ShutdownError::Irpc { 804 804 + message: e.to_string(), 805 805 + }) 806 806 + } 807 807 + }

+2 -3

rust/iroh-streamplace/src/node/db.rs rust/iroh-streamplace/src/db.rs

··· 12 12 use snafu::Snafu; 13 13 use tokio::sync::Mutex; 14 14 15 15 - use super::db; 16 15 use crate::public_key::PublicKey; 17 16 18 17 // the files here are just copied from iroh-smol-kv-uniffi/src/code ··· 22 21 mod subscribe_mode; 23 22 pub use subscribe_mode::SubscribeMode; 24 23 } 25 25 - use db::util::format_bytes; 26 24 pub use kv::{SubscribeMode, TimeBound}; 25 25 + use util::format_bytes; 27 26 28 27 /// Error creating a new database node. 29 28 #[derive(Debug, Snafu, uniffi::Error)] ··· 542 541 543 542 #[cfg(test)] 544 543 mod tests { 545 545 - use super::db::util; 544 544 + use super::util; 546 545 547 546 #[test] 548 547 fn escape_unescape() {

rust/iroh-streamplace/src/node/db/kv/subscribe_mode.rs rust/iroh-streamplace/src/db/kv/subscribe_mode.rs

rust/iroh-streamplace/src/node/db/kv/time_bound.rs rust/iroh-streamplace/src/db/kv/time_bound.rs

-13

rust/iroh-streamplace/src/node/mod.rs

··· 1 1 - use std::sync::LazyLock; 2 2 - 3 3 - /// We export the entire API at top level since this is what go-uniffi-bindgen will do anyway. 4 4 - mod streams; 5 5 - pub use streams::*; 6 6 - mod db; 7 7 - pub use db::*; 8 8 - #[cfg(test)] 9 9 - mod tests; 10 10 - 11 11 - /// Lazily initialized Tokio runtime for use in uniffi methods that need a runtime. 12 12 - static RUNTIME: LazyLock<tokio::runtime::Runtime> = 13 13 - LazyLock::new(|| tokio::runtime::Runtime::new().unwrap());

-809

rust/iroh-streamplace/src/node/streams.rs

··· 1 1 - use std::{ 2 2 - collections::{BTreeMap, BTreeSet, HashSet}, 3 3 - str::FromStr, 4 4 - sync::Arc, 5 5 - }; 6 6 - 7 7 - use bytes::Bytes; 8 8 - use iroh::{NodeId, PublicKey, RelayMode, SecretKey, Watcher}; 9 9 - use iroh_base::ticket::NodeTicket; 10 10 - use iroh_gossip::{net::Gossip, proto::TopicId}; 11 11 - use irpc::{WithChannels, rpc::RemoteService}; 12 12 - use irpc_iroh::{IrohProtocol, IrohRemoteConnection}; 13 13 - use n0_future::{TryFutureExt, future::Boxed}; 14 14 - 15 15 - mod rpc { 16 16 - //! Protocol API 17 17 - use bytes::Bytes; 18 18 - use iroh::NodeId; 19 19 - use irpc::{channel::oneshot, rpc_requests}; 20 20 - use serde::{Deserialize, Serialize}; 21 21 - 22 22 - pub const ALPN: &[u8] = b"/iroh/streamplace/1"; 23 23 - 24 24 - /// Subscribe to the given `key` 25 25 - #[derive(Debug, Serialize, Deserialize)] 26 26 - pub struct Subscribe { 27 27 - pub key: String, 28 28 - // TODO: verify 29 29 - pub remote_id: NodeId, 30 30 - } 31 31 - 32 32 - /// Unsubscribe from the given `key` 33 33 - #[derive(Debug, Serialize, Deserialize)] 34 34 - pub struct Unsubscribe { 35 35 - pub key: String, 36 36 - // TODO: verify 37 37 - pub remote_id: NodeId, 38 38 - } 39 39 - 40 40 - // #[derive(Debug, Serialize, Deserialize)] 41 41 - // pub struct SendSegment { 42 42 - // pub key: String, 43 43 - // pub data: Bytes, 44 44 - // } 45 45 - 46 46 - #[derive(Debug, Clone, Serialize, Deserialize)] 47 47 - pub struct RecvSegment { 48 48 - pub key: String, 49 49 - pub data: Bytes, 50 50 - } 51 51 - 52 52 - // Use the macro to generate both the Protocol and Message enums 53 53 - // plus implement Channels for each type 54 54 - #[rpc_requests(message = Message)] 55 55 - #[derive(Serialize, Deserialize, Debug)] 56 56 - pub enum Protocol { 57 57 - #[rpc(tx=oneshot::Sender<()>)] 58 58 - Subscribe(Subscribe), 59 59 - #[rpc(tx=oneshot::Sender<()>)] 60 60 - Unsubscribe(Unsubscribe), 61 61 - #[rpc(tx=oneshot::Sender<()>)] 62 62 - RecvSegment(RecvSegment), 63 63 - } 64 64 - } 65 65 - 66 66 - mod api { 67 67 - //! Protocol API 68 68 - use bytes::Bytes; 69 69 - use iroh::{NodeAddr, NodeId}; 70 70 - use irpc::{channel::oneshot, rpc_requests}; 71 71 - use serde::{Deserialize, Serialize}; 72 72 - 73 73 - /// Subscribe to the given `key` 74 74 - #[derive(Debug, Serialize, Deserialize)] 75 75 - pub struct Subscribe { 76 76 - pub key: String, 77 77 - // TODO: verify 78 78 - pub remote_id: NodeId, 79 79 - } 80 80 - 81 81 - /// Unsubscribe from the given `key` 82 82 - #[derive(Debug, Serialize, Deserialize)] 83 83 - pub struct Unsubscribe { 84 84 - pub key: String, 85 85 - // TODO: verify 86 86 - pub remote_id: NodeId, 87 87 - } 88 88 - 89 89 - #[derive(Debug, Serialize, Deserialize)] 90 90 - pub struct SendSegment { 91 91 - pub key: String, 92 92 - pub data: Bytes, 93 93 - } 94 94 - 95 95 - #[derive(Debug, Serialize, Deserialize)] 96 96 - pub struct JoinPeers { 97 97 - pub peers: Vec<NodeAddr>, 98 98 - } 99 99 - 100 100 - #[derive(Debug, Serialize, Deserialize)] 101 101 - pub struct AddTickets { 102 102 - pub peers: Vec<NodeAddr>, 103 103 - } 104 104 - 105 105 - #[derive(Debug, Serialize, Deserialize)] 106 106 - pub struct GetNodeAddr; 107 107 - 108 108 - #[derive(Debug, Serialize, Deserialize)] 109 109 - pub struct Shutdown; 110 110 - 111 111 - // Use the macro to generate both the Protocol and Message enums 112 112 - // plus implement Channels for each type 113 113 - #[rpc_requests(message = Message)] 114 114 - #[derive(Serialize, Deserialize, Debug)] 115 115 - pub enum Protocol { 116 116 - #[rpc(tx=oneshot::Sender<()>)] 117 117 - Subscribe(Subscribe), 118 118 - #[rpc(tx=oneshot::Sender<()>)] 119 119 - Unsubscribe(Unsubscribe), 120 120 - #[rpc(tx=oneshot::Sender<()>)] 121 121 - SendSegment(SendSegment), 122 122 - #[rpc(tx=oneshot::Sender<()>)] 123 123 - JoinPeers(JoinPeers), 124 124 - #[rpc(tx=oneshot::Sender<()>)] 125 125 - AddTickets(AddTickets), 126 126 - #[rpc(tx=oneshot::Sender<NodeAddr>)] 127 127 - GetNodeAddr(GetNodeAddr), 128 128 - #[rpc(tx=oneshot::Sender<()>)] 129 129 - Shutdown(Shutdown), 130 130 - } 131 131 - } 132 132 - use api::{Message as ApiMessage, Protocol as ApiProtocol}; 133 133 - use n0_future::{FuturesUnordered, StreamExt}; 134 134 - use rpc::{Message as RpcMessage, Protocol as RpcProtocol}; 135 135 - use snafu::Snafu; 136 136 - use tokio::sync::oneshot; 137 137 - use tracing::{Instrument, debug, error, trace, trace_span, warn}; 138 138 - 139 139 - use super::{Config, CreateError, JoinPeersError, PutError, db, streams::rpc::RecvSegment}; 140 140 - use crate::node::ShutdownError; 141 141 - 142 142 - pub(crate) enum HandlerMode { 143 143 - Sender, 144 144 - Forwarder, 145 145 - Receiver(Box<dyn Fn(String, Vec<u8>) -> Boxed<()> + Send + Sync + 'static>), 146 146 - } 147 147 - 148 148 - impl HandlerMode { 149 149 - pub fn receiver_fn<F, Fut>(f: F) -> Self 150 150 - where 151 151 - F: Fn(String, Vec<u8>) -> Fut + Send + Sync + 'static, 152 152 - Fut: std::future::Future<Output = ()> + Send + 'static, 153 153 - { 154 154 - Self::Receiver(Box::new(move |name, data| Box::pin(f(name, data)))) 155 155 - } 156 156 - 157 157 - pub fn receiver(handler: Arc<dyn DataHandler>) -> Self { 158 158 - Self::receiver_fn(move |id, data| { 159 159 - let handler = handler.clone(); 160 160 - async move { 161 161 - handler.handle_data(id, data).await; 162 162 - } 163 163 - }) 164 164 - } 165 165 - 166 166 - pub fn mode_str(&self) -> &'static str { 167 167 - match self { 168 168 - HandlerMode::Sender => "sender", 169 169 - HandlerMode::Forwarder => "forwarder", 170 170 - HandlerMode::Receiver(_) => "receiver", 171 171 - } 172 172 - } 173 173 - } 174 174 - 175 175 - type Tasks = FuturesUnordered<Boxed<(NodeId, Result<(), RpcTaskError>)>>; 176 176 - 177 177 - /// Actor that contains both a kv db for metadata and a handler for the rpc protocol. 178 178 - /// 179 179 - /// This can be used both for sender and receiver nodes. Sender nodes will just set the 180 180 - /// handler to None. 181 181 - struct Actor { 182 182 - /// Receiver for rpc messages from remote nodes 183 183 - rpc_rx: tokio::sync::mpsc::Receiver<RpcMessage>, 184 184 - /// Receiver for API messages from the user 185 185 - api_rx: tokio::sync::mpsc::Receiver<ApiMessage>, 186 186 - /// nodes I need to send to for each stream 187 187 - subscribers: BTreeMap<String, BTreeSet<NodeId>>, 188 188 - /// nodes I am subscribed to 189 189 - subscriptions: BTreeMap<String, NodeId>, 190 190 - /// lightweight typed connection pool 191 191 - connections: ConnectionPool, 192 192 - /// How to handle incoming data 193 193 - handler: HandlerMode, 194 194 - /// Iroh protocol router, I need to keep it around to keep the protocol alive 195 195 - router: iroh::protocol::Router, 196 196 - /// Metadata db 197 197 - client: db::Db, 198 198 - /// Write scope for this node for the metadata db 199 199 - write: db::WriteScope, 200 200 - /// Ongoing tasks 201 201 - tasks: Tasks, 202 202 - /// Configuration, needed for timeouts etc. 203 203 - config: Arc<super::Config>, 204 204 - } 205 205 - 206 206 - #[derive(Debug, Clone)] 207 207 - struct Connection { 208 208 - id: NodeId, 209 209 - rpc: irpc::Client<RpcProtocol>, 210 210 - } 211 211 - 212 212 - #[derive(Debug, Snafu)] 213 213 - enum RpcTaskError { 214 214 - #[snafu(transparent)] 215 215 - Task { source: irpc::Error }, 216 216 - #[snafu(transparent)] 217 217 - Timeout { source: tokio::time::error::Elapsed }, 218 218 - } 219 219 - 220 220 - struct ConnectionPool { 221 221 - endpoint: iroh::Endpoint, 222 222 - connections: BTreeMap<NodeId, Connection>, 223 223 - } 224 224 - 225 225 - impl ConnectionPool { 226 226 - fn new(endpoint: iroh::Endpoint) -> Self { 227 227 - Self { 228 228 - endpoint, 229 229 - connections: BTreeMap::new(), 230 230 - } 231 231 - } 232 232 - 233 233 - /// Cheap conn pool hack 234 234 - fn get(&mut self, remote: &NodeId) -> Connection { 235 235 - if !self.connections.contains_key(remote) { 236 236 - let conn = IrohRemoteConnection::new( 237 237 - self.endpoint.clone(), 238 238 - (*remote).into(), 239 239 - rpc::ALPN.to_vec(), 240 240 - ); 241 241 - let conn = Connection { 242 242 - rpc: irpc::Client::boxed(conn), 243 243 - id: *remote, 244 244 - }; 245 245 - self.connections.insert(*remote, conn); 246 246 - } 247 247 - self.connections 248 248 - .get_mut(remote) 249 249 - .expect("just inserted") 250 250 - .clone() 251 251 - } 252 252 - 253 253 - fn remove(&mut self, remote: &NodeId) { 254 254 - self.connections.remove(remote); 255 255 - } 256 256 - } 257 257 - 258 258 - impl Actor { 259 259 - pub async fn spawn( 260 260 - endpoint: iroh::Endpoint, 261 261 - topic: iroh_gossip::proto::TopicId, 262 262 - config: super::Config, 263 263 - handler: HandlerMode, 264 264 - ) -> Result<(Node, impl Future<Output = ()>), iroh_gossip::api::ApiError> { 265 265 - let (rpc_tx, rpc_rx) = tokio::sync::mpsc::channel::<RpcMessage>(32); 266 266 - let (api_tx, api_rx) = tokio::sync::mpsc::channel::<ApiMessage>(32); 267 267 - let gossip = Gossip::builder().spawn(endpoint.clone()); 268 268 - let id = endpoint.node_id(); 269 269 - let router = iroh::protocol::Router::builder(endpoint.clone()) 270 270 - .accept(iroh_gossip::ALPN, gossip.clone()) 271 271 - .accept( 272 272 - rpc::ALPN, 273 273 - IrohProtocol::new(rpc::Protocol::remote_handler(rpc_tx.into())), 274 274 - ) 275 275 - .spawn(); 276 276 - let topic = gossip.subscribe(topic, vec![]).await?; 277 277 - let secret = router.endpoint().secret_key().clone(); 278 278 - let db_config = Default::default(); 279 279 - let client = iroh_smol_kv::Client::local(topic, db_config); 280 280 - let write = db::WriteScope::new(client.write(secret.clone())); 281 281 - let client = db::Db::new(client); 282 282 - let actor = Self { 283 283 - rpc_rx, 284 284 - api_rx, 285 285 - subscribers: BTreeMap::new(), 286 286 - subscriptions: BTreeMap::new(), 287 287 - connections: ConnectionPool::new(router.endpoint().clone()), 288 288 - handler, 289 289 - router, 290 290 - write: write.clone(), 291 291 - client: client.clone(), 292 292 - tasks: FuturesUnordered::new(), 293 293 - config: Arc::new(config), 294 294 - }; 295 295 - let api = Node { 296 296 - client: Arc::new(client), 297 297 - write: Arc::new(write), 298 298 - api: irpc::Client::local(api_tx), 299 299 - }; 300 300 - Ok(( 301 301 - api, 302 302 - actor 303 303 - .run() 304 304 - .instrument(trace_span!("actor", id=%id.fmt_short())), 305 305 - )) 306 306 - } 307 307 - 308 308 - async fn run(mut self) { 309 309 - loop { 310 310 - tokio::select! { 311 311 - msg = self.rpc_rx.recv() => { 312 312 - let Some(msg) = msg else { 313 313 - error!("rpc channel closed"); 314 314 - break; 315 315 - }; 316 316 - self.handle_rpc(msg).instrument(trace_span!("rpc")).await; 317 317 - } 318 318 - msg = self.api_rx.recv() => { 319 319 - let Some(msg) = msg else { 320 320 - break; 321 321 - }; 322 322 - if let Some(shutdown) = self.handle_api(msg).instrument(trace_span!("api")).await { 323 323 - shutdown.send(()).await.ok(); 324 324 - break; 325 325 - } 326 326 - } 327 327 - res = self.tasks.next(), if !self.tasks.is_empty() => { 328 328 - let Some((remote_id, res)) = res else { 329 329 - error!("task finished but no result"); 330 330 - break; 331 331 - }; 332 332 - match res { 333 333 - Ok(()) => {} 334 334 - Err(RpcTaskError::Timeout { source }) => { 335 335 - warn!("call to {remote_id} timed out: {source}"); 336 336 - } 337 337 - Err(RpcTaskError::Task { source }) => { 338 338 - warn!("call to {remote_id} failed: {source}"); 339 339 - } 340 340 - } 341 341 - self.connections.remove(&remote_id); 342 342 - } 343 343 - } 344 344 - } 345 345 - } 346 346 - 347 347 - async fn update_subscriber_meta(&mut self, key: &str) { 348 348 - let n = self 349 349 - .subscribers 350 350 - .get(key) 351 351 - .map(|s| s.len()) 352 352 - .unwrap_or_default(); 353 353 - let v = n.to_string().into_bytes(); 354 354 - self.write 355 355 - .put_impl(Some(key.as_bytes().to_vec()), b"subscribers", v.into()) 356 356 - .await 357 357 - .ok(); 358 358 - } 359 359 - 360 360 - /// Requests from remote nodes 361 361 - async fn handle_rpc(&mut self, msg: RpcMessage) { 362 362 - match msg { 363 363 - RpcMessage::Subscribe(msg) => { 364 364 - trace!("{:?}", msg.inner); 365 365 - let WithChannels { 366 366 - tx, 367 367 - inner: rpc::Subscribe { key, remote_id }, 368 368 - .. 369 369 - } = msg; 370 370 - self.subscribers 371 371 - .entry(key.clone()) 372 372 - .or_default() 373 373 - .insert(remote_id); 374 374 - self.update_subscriber_meta(&key).await; 375 375 - tx.send(()).await.ok(); 376 376 - } 377 377 - RpcMessage::Unsubscribe(msg) => { 378 378 - debug!("{:?}", msg.inner); 379 379 - let WithChannels { 380 380 - tx, 381 381 - inner: rpc::Unsubscribe { key, remote_id }, 382 382 - .. 383 383 - } = msg; 384 384 - if let Some(e) = self.subscribers.get_mut(&key) 385 385 - && !e.remove(&remote_id) 386 386 - { 387 387 - warn!( 388 388 - "unsubscribe: no subscription for {} from {}", 389 389 - key, remote_id 390 390 - ); 391 391 - } 392 392 - if let Some(subscriptions) = self.subscribers.get(&key) 393 393 - && subscriptions.is_empty() 394 394 - { 395 395 - self.subscribers.remove(&key); 396 396 - } 397 397 - self.update_subscriber_meta(&key).await; 398 398 - tx.send(()).await.ok(); 399 399 - } 400 400 - RpcMessage::RecvSegment(msg) => { 401 401 - trace!("{:?}", msg.inner); 402 402 - let WithChannels { 403 403 - tx, 404 404 - inner: rpc::RecvSegment { key, data }, 405 405 - .. 406 406 - } = msg; 407 407 - match &self.handler { 408 408 - HandlerMode::Sender => { 409 409 - warn!("received segment but in sender mode"); 410 410 - } 411 411 - HandlerMode::Forwarder => { 412 412 - if let Some(remotes) = self.subscribers.get(&key) { 413 413 - Self::handle_send( 414 414 - &mut self.tasks, 415 415 - &mut self.connections, 416 416 - &self.config, 417 417 - key, 418 418 - data, 419 419 - remotes, 420 420 - ); 421 421 - } else { 422 422 - trace!("no subscribers for stream {}", key); 423 423 - } 424 424 - } 425 425 - HandlerMode::Receiver(handler) => { 426 426 - if self.subscriptions.contains_key(&key) { 427 427 - handler(key, data.to_vec()).await; 428 428 - } else { 429 429 - warn!("received segment for unsubscribed key: {}", key); 430 430 - } 431 431 - } 432 432 - }; 433 433 - tx.send(()).await.ok(); 434 434 - } 435 435 - } 436 436 - } 437 437 - 438 438 - async fn handle_api(&mut self, msg: ApiMessage) -> Option<irpc::channel::oneshot::Sender<()>> { 439 439 - match msg { 440 440 - ApiMessage::SendSegment(msg) => { 441 441 - trace!("{:?}", msg.inner); 442 442 - let WithChannels { 443 443 - tx, 444 444 - inner: api::SendSegment { key, data }, 445 445 - .. 446 446 - } = msg; 447 447 - if let Some(remotes) = self.subscribers.get(&key) { 448 448 - Self::handle_send( 449 449 - &mut self.tasks, 450 450 - &mut self.connections, 451 451 - &self.config, 452 452 - key, 453 453 - data, 454 454 - remotes, 455 455 - ); 456 456 - } else { 457 457 - trace!("no subscribers for stream {}", key); 458 458 - } 459 459 - tx.send(()).await.ok(); 460 460 - } 461 461 - ApiMessage::Subscribe(msg) => { 462 462 - trace!("{:?}", msg.inner); 463 463 - let WithChannels { 464 464 - tx, 465 465 - inner: api::Subscribe { key, remote_id }, 466 466 - .. 467 467 - } = msg; 468 468 - let conn = self.connections.get(&remote_id); 469 469 - conn.rpc 470 470 - .rpc(rpc::Subscribe { 471 471 - key: key.clone(), 472 472 - remote_id: self.node_id(), 473 473 - }) 474 474 - .await 475 475 - .ok(); 476 476 - self.subscriptions.insert(key, remote_id); 477 477 - tx.send(()).await.ok(); 478 478 - } 479 479 - ApiMessage::Unsubscribe(msg) => { 480 480 - trace!("{:?}", msg.inner); 481 481 - let WithChannels { 482 482 - tx, 483 483 - inner: api::Unsubscribe { key, remote_id }, 484 484 - .. 485 485 - } = msg; 486 486 - let conn = self.connections.get(&remote_id); 487 487 - conn.rpc 488 488 - .rpc(rpc::Unsubscribe { 489 489 - key: key.clone(), 490 490 - remote_id: self.node_id(), 491 491 - }) 492 492 - .await 493 493 - .ok(); 494 494 - self.subscriptions.remove(&key); 495 495 - tx.send(()).await.ok(); 496 496 - } 497 497 - ApiMessage::AddTickets(msg) => { 498 498 - trace!("{:?}", msg.inner); 499 499 - let WithChannels { 500 500 - tx, 501 501 - inner: api::AddTickets { peers }, 502 502 - .. 503 503 - } = msg; 504 504 - for addr in &peers { 505 505 - self.router.endpoint().add_node_addr(addr.clone()).ok(); 506 506 - } 507 507 - // self.client.inner().join_peers(ids).await.ok(); 508 508 - tx.send(()).await.ok(); 509 509 - } 510 510 - ApiMessage::JoinPeers(msg) => { 511 511 - trace!("{:?}", msg.inner); 512 512 - let WithChannels { 513 513 - tx, 514 514 - inner: api::JoinPeers { peers }, 515 515 - .. 516 516 - } = msg; 517 517 - let ids = peers 518 518 - .iter() 519 519 - .map(|a| a.node_id) 520 520 - .filter(|id| *id != self.node_id()) 521 521 - .collect::<HashSet<_>>(); 522 522 - for addr in &peers { 523 523 - self.router.endpoint().add_node_addr(addr.clone()).ok(); 524 524 - } 525 525 - self.client.inner().join_peers(ids).await.ok(); 526 526 - tx.send(()).await.ok(); 527 527 - } 528 528 - ApiMessage::GetNodeAddr(msg) => { 529 529 - trace!("{:?}", msg.inner); 530 530 - let WithChannels { tx, .. } = msg; 531 531 - if !self.config.disable_relay { 532 532 - // don't await home relay if we have disabled relays, this will hang forever 533 533 - self.router.endpoint().home_relay().initialized().await; 534 534 - } 535 535 - let addr = self.router.endpoint().node_addr().initialized().await; 536 536 - tx.send(addr).await.ok(); 537 537 - } 538 538 - ApiMessage::Shutdown(msg) => { 539 539 - return Some(msg.tx); 540 540 - } 541 541 - } 542 542 - None 543 543 - } 544 544 - 545 545 - fn handle_send( 546 546 - tasks: &mut Tasks, 547 547 - connections: &mut ConnectionPool, 548 548 - config: &Arc<Config>, 549 549 - key: String, 550 550 - data: Bytes, 551 551 - remotes: &BTreeSet<NodeId>, 552 552 - ) { 553 553 - let msg = rpc::RecvSegment { key, data }; 554 554 - for remote in remotes { 555 555 - trace!("sending to stream {}: {}", msg.key, remote); 556 556 - let conn = connections.get(remote); 557 557 - tasks.push(Box::pin(Self::forward_task( 558 558 - config.clone(), 559 559 - conn, 560 560 - msg.clone(), 561 561 - ))); 562 562 - } 563 563 - } 564 564 - 565 565 - async fn forward_task( 566 566 - config: Arc<super::Config>, 567 567 - conn: Connection, 568 568 - msg: RecvSegment, 569 569 - ) -> (NodeId, Result<(), RpcTaskError>) { 570 570 - let id = conn.id; 571 571 - let res = async move { 572 572 - tokio::time::timeout(config.max_send_duration, conn.rpc.rpc(msg)).await??; 573 573 - Ok(()) 574 574 - } 575 575 - .await; 576 576 - (id, res) 577 577 - } 578 578 - 579 579 - fn node_id(&self) -> PublicKey { 580 580 - self.router.endpoint().node_id() 581 581 - } 582 582 - } 583 583 - 584 584 - /// Iroh-streamplace node that can send, forward or receive stream segments. 585 585 - #[derive(Clone, uniffi::Object)] 586 586 - pub struct Node { 587 587 - client: Arc<db::Db>, 588 588 - write: Arc<db::WriteScope>, 589 589 - api: irpc::Client<ApiProtocol>, 590 590 - } 591 591 - 592 592 - impl Node { 593 593 - pub(crate) async fn new_in_runtime( 594 594 - config: super::Config, 595 595 - handler: HandlerMode, 596 596 - ) -> Result<Arc<Self>, CreateError> { 597 597 - let mode_str = Bytes::from(handler.mode_str()); 598 598 - let secret_key = 599 599 - SecretKey::from_bytes(&<[u8; 32]>::try_from(config.key.clone()).map_err(|e| { 600 600 - CreateError::PrivateKey { 601 601 - size: e.len() as u64, 602 602 - } 603 603 - })?); 604 604 - let topic = 605 605 - TopicId::from_bytes(<[u8; 32]>::try_from(config.topic.clone()).map_err(|e| { 606 606 - CreateError::Topic { 607 607 - size: e.len() as u64, 608 608 - } 609 609 - })?); 610 610 - let relay_mode = if config.disable_relay { 611 611 - RelayMode::Disabled 612 612 - } else { 613 613 - RelayMode::Default 614 614 - }; 615 615 - let endpoint = iroh::Endpoint::builder() 616 616 - .secret_key(secret_key) 617 617 - .relay_mode(relay_mode) 618 618 - .bind() 619 619 - .await 620 620 - .map_err(|e| CreateError::Bind { 621 621 - message: e.to_string(), 622 622 - })?; 623 623 - let (api, actor) = Actor::spawn(endpoint, topic, config, handler) 624 624 - .await 625 625 - .map_err(|e| CreateError::Subscribe { 626 626 - message: e.to_string(), 627 627 - })?; 628 628 - api.node_scope() 629 629 - .put_impl(Option::<Vec<u8>>::None, b"mode", mode_str) 630 630 - .await 631 631 - .ok(); 632 632 - tokio::spawn(actor); 633 633 - Ok(Arc::new(api)) 634 634 - } 635 635 - } 636 636 - 637 637 - /// DataHandler trait that is exported to go for receiving data callbacks. 638 638 - #[uniffi::export(with_foreign)] 639 639 - #[async_trait::async_trait] 640 640 - pub trait DataHandler: Send + Sync { 641 641 - async fn handle_data(&self, topic: String, data: Vec<u8>); 642 642 - } 643 643 - 644 644 - #[uniffi::export] 645 645 - impl Node { 646 646 - /// Create a new streamplace client node. 647 647 - #[uniffi::constructor] 648 648 - pub async fn sender(config: super::Config) -> Result<Arc<Self>, CreateError> { 649 649 - super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Sender)) 650 650 - } 651 651 - 652 652 - #[uniffi::constructor] 653 653 - pub async fn forwarder(config: super::Config) -> Result<Arc<Self>, CreateError> { 654 654 - super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Forwarder)) 655 655 - } 656 656 - 657 657 - #[uniffi::constructor] 658 658 - pub async fn receiver( 659 659 - config: super::Config, 660 660 - handler: Arc<dyn DataHandler>, 661 661 - ) -> Result<Arc<Self>, CreateError> { 662 662 - super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::receiver(handler))) 663 663 - } 664 664 - 665 665 - /// Get a handle to the db to watch for changes locally or globally. 666 666 - pub fn db(&self) -> Arc<db::Db> { 667 667 - self.client.clone() 668 668 - } 669 669 - 670 670 - /// Get a handle to the write scope for this node. 671 671 - /// 672 672 - /// This is equivalent to calling `db.write(...)` with the secret key used to create the node. 673 673 - pub fn node_scope(&self) -> Arc<db::WriteScope> { 674 674 - self.write.clone() 675 675 - } 676 676 - 677 677 - /// Subscribe to updates for a given stream from a remote node. 678 678 - pub async fn subscribe( 679 679 - &self, 680 680 - key: String, 681 681 - remote_id: Arc<crate::public_key::PublicKey>, 682 682 - ) -> Result<(), PutError> { 683 683 - self.api 684 684 - .rpc(api::Subscribe { 685 685 - key, 686 686 - remote_id: remote_id.as_ref().into(), 687 687 - }) 688 688 - .await 689 689 - .map_err(|e| PutError::Irpc { 690 690 - message: e.to_string(), 691 691 - }) 692 692 - } 693 693 - 694 694 - /// Unsubscribe from updates for a given stream from a remote node. 695 695 - pub async fn unsubscribe( 696 696 - &self, 697 697 - key: String, 698 698 - remote_id: Arc<crate::public_key::PublicKey>, 699 699 - ) -> Result<(), PutError> { 700 700 - self.api 701 701 - .rpc(api::Unsubscribe { 702 702 - key, 703 703 - remote_id: remote_id.as_ref().into(), 704 704 - }) 705 705 - .await 706 706 - .map_err(|e| PutError::Irpc { 707 707 - message: e.to_string(), 708 708 - }) 709 709 - } 710 710 - 711 711 - /// Send a segment to all subscribers of the given stream. 712 712 - pub async fn send_segment(&self, key: String, data: Vec<u8>) -> Result<(), PutError> { 713 713 - self.api 714 714 - .rpc(api::SendSegment { 715 715 - key, 716 716 - data: data.into(), 717 717 - }) 718 718 - .await 719 719 - .map_err(|e| PutError::Irpc { 720 720 - message: e.to_string(), 721 721 - }) 722 722 - } 723 723 - 724 724 - /// Join peers by their node tickets. 725 725 - pub async fn join_peers(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { 726 726 - let peers = peers 727 727 - .iter() 728 728 - .map(|p| NodeTicket::from_str(p)) 729 729 - .collect::<Result<Vec<_>, _>>() 730 730 - .map_err(|e| JoinPeersError::Ticket { 731 731 - message: e.to_string(), 732 732 - })?; 733 733 - let addrs = peers 734 734 - .iter() 735 735 - .map(|t| t.node_addr().clone()) 736 736 - .collect::<Vec<_>>(); 737 737 - self.api 738 738 - .rpc(api::JoinPeers { peers: addrs }) 739 739 - .await 740 740 - .map_err(|e| JoinPeersError::Irpc { 741 741 - message: e.to_string(), 742 742 - }) 743 743 - } 744 744 - 745 745 - /// Add tickets for remote peers 746 746 - pub async fn add_tickets(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { 747 747 - let peers = peers 748 748 - .iter() 749 749 - .map(|p| NodeTicket::from_str(p)) 750 750 - .collect::<Result<Vec<_>, _>>() 751 751 - .map_err(|e| JoinPeersError::Ticket { 752 752 - message: e.to_string(), 753 753 - })?; 754 754 - let addrs = peers 755 755 - .iter() 756 756 - .map(|t| t.node_addr().clone()) 757 757 - .collect::<Vec<_>>(); 758 758 - self.api 759 759 - .rpc(api::AddTickets { peers: addrs }) 760 760 - .await 761 761 - .map_err(|e| JoinPeersError::Irpc { 762 762 - message: e.to_string(), 763 763 - }) 764 764 - } 765 765 - 766 766 - /// Get this node's ticket. 767 767 - pub async fn ticket(&self) -> Result<String, PutError> { 768 768 - let addr = self 769 769 - .api 770 770 - .rpc(api::GetNodeAddr) 771 771 - .await 772 772 - .map_err(|e| PutError::Irpc { 773 773 - message: e.to_string(), 774 774 - })?; 775 775 - Ok(NodeTicket::from(addr).to_string()) 776 776 - } 777 777 - 778 778 - /// Get this node's node ID. 779 779 - pub async fn node_id(&self) -> Result<Arc<crate::public_key::PublicKey>, PutError> { 780 780 - let addr = self 781 781 - .api 782 782 - .rpc(api::GetNodeAddr) 783 783 - .await 784 784 - .map_err(|e| PutError::Irpc { 785 785 - message: e.to_string(), 786 786 - })?; 787 787 - Ok(Arc::new(addr.node_id.into())) 788 788 - } 789 789 - 790 790 - /// Shutdown the node, including the streaming system and the metadata db. 791 791 - pub async fn shutdown(&self) -> Result<(), ShutdownError> { 792 792 - // shut down both the streams and the db concurrently, even if one fails 793 793 - let (res1, res2) = tokio::join!(self.shutdown_streams(), self.client.shutdown()); 794 794 - res1?; 795 795 - res2?; 796 796 - Ok(()) 797 797 - } 798 798 - } 799 799 - 800 800 - impl Node { 801 801 - async fn shutdown_streams(&self) -> std::result::Result<(), ShutdownError> { 802 802 - self.api 803 803 - .rpc(api::Shutdown) 804 804 - .await 805 805 - .map_err(|e| ShutdownError::Irpc { 806 806 - message: e.to_string(), 807 807 - }) 808 808 - } 809 809 - }

+5 -5

rust/iroh-streamplace/src/node/tests.rs rust/iroh-streamplace/src/tests.rs

··· 4 4 use n0_future::{BufferedStreamExt, StreamExt, stream}; 5 5 use testresult::TestResult; 6 6 7 7 - use super::{streams::HandlerMode, *}; 7 7 + use super::*; 8 8 9 9 struct TestNode { 10 10 node: Arc<Node>, ··· 144 144 let (tx, mut rx) = tokio::sync::mpsc::channel(32); 145 145 let handler = Arc::new(TestHandler::new((), tx)); 146 146 let sender = TestNode::new(HandlerMode::Sender).await?; 147 147 - let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; 147 147 + let receiver = TestNode::new(HandlerMode::Receiver(handler)).await?; 148 148 // join the sender to the receiver. This will also configure the receiver endpoint to be able to dial the sender. 149 149 receiver.join_peers(vec![sender.ticket.clone()]).await?; 150 150 let stream = "teststream".to_string(); ··· 165 165 let handler = Arc::new(TestHandler::new((), tx)); 166 166 let sender = TestNode::new(HandlerMode::Sender).await?; 167 167 let forwarder = TestNode::new(HandlerMode::Forwarder).await?; 168 168 - let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; 168 168 + let receiver = TestNode::new(HandlerMode::Receiver(handler)).await?; 169 169 // join everyone to everyone, so the receiver can reach the sender via the forwarder. 170 170 let tickets = vec![ 171 171 sender.ticket.clone(), ··· 196 196 let handler = Arc::new(TestHandler::new((), tx)); 197 197 let sender = TestNode::new(HandlerMode::Sender).await?; 198 198 let forwarder = TestNode::new(HandlerMode::Forwarder).await?; 199 199 - let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; 199 199 + let receiver = TestNode::new(HandlerMode::Receiver(handler)).await?; 200 200 // join everyone to everyone, so the receiver can reach the sender via the forwarder. 201 201 let tickets = vec![ 202 202 sender.ticket.clone(), ··· 241 241 } else if forwarders.contains(&i) { 242 242 HandlerMode::Forwarder 243 243 } else { 244 244 - HandlerMode::receiver(Arc::new(TestHandler::new(i, tx.clone()))) 244 244 + HandlerMode::Receiver(Arc::new(TestHandler::new(i, tx.clone()))) 245 245 } 246 246 }; 247 247 let nodes = test_nodes(ntotal, make_handler, true).await?;

rust/iroh-streamplace/src/streams.rs

This is a binary file and will not be displayed.