diff --git a/pkg/aqio/aqio.go b/pkg/aqio/aqio.go index 3c3824a18..4979eb2c4 100644 --- a/pkg/aqio/aqio.go +++ b/pkg/aqio/aqio.go @@ -2,7 +2,6 @@ package aqio import ( "errors" - "fmt" "io" "github.com/johncgriffin/overflow" @@ -19,11 +18,13 @@ type ReadWriteSeeker struct { } // Write implements the io.Writer interface +// TODO: This would probably be better as a linked list for writing; the read +// requirements are minimal so doing a bit more math is okay func (rws *ReadWriteSeeker) Write(p []byte) (n int, err error) { - fmt.Printf("Write: pos=%d len(p)=%d\n", rws.pos, len(p)) + // fmt.Printf("Write: pos=%d len(p)=%d\n", rws.pos, len(p)) minCap := overflow.Addp(rws.pos, len(p)) if minCap > cap(rws.buf) { // Make sure buf has enough capacity: - fmt.Printf("Write: pos=%d len(p)=%d minCap=%d\n", rws.pos, len(p), minCap) + // fmt.Printf("Write: pos=%d len(p)=%d minCap=%d\n", rws.pos, len(p), minCap) newCap := cap(rws.buf) * 2 if newCap == 0 { newCap = 128 diff --git a/pkg/c2patypes/stream_adapter.go b/pkg/c2patypes/stream_adapter.go index c1d8edb07..33017bea9 100644 --- a/pkg/c2patypes/stream_adapter.go +++ b/pkg/c2patypes/stream_adapter.go @@ -4,6 +4,8 @@ import ( "errors" "fmt" "io" + + "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" ) // type Stream interface { @@ -121,3 +123,31 @@ func writeStream(w io.ReadWriteSeeker, data []byte) (uint64, error) { } return uint64(wrote), nil } + +type ManyStreams struct { + Streams []io.ReadSeeker + index int +} + +func NewManyStreams() *ManyStreams { + return &ManyStreams{Streams: []io.ReadSeeker{}, index: 0} +} + +func (m *ManyStreams) AddStream(stream io.ReadSeeker) { + m.Streams = append(m.Streams, stream) +} + +func (m *ManyStreams) Next() (*iroh_streamplace.Stream, error) { + if m.index >= len(m.Streams) { + return nil, nil + } + stream := m.Streams[m.index] + m.index++ + var r iroh_streamplace.Stream + if rws, ok := stream.(io.ReadWriteSeeker); ok { + r = NewWriter(rws) + } else { + r = NewReader(stream) + } + return &r, nil +} diff --git a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go index 8312662ff..e1e9adf5f 100644 --- a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go +++ b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go @@ -338,6 +338,7 @@ func init() { FfiConverterDataHandlerINSTANCE.register() FfiConverterGoSignerINSTANCE.register() + FfiConverterManyStreamsINSTANCE.register() FfiConverterStreamINSTANCE.register() uniffiCheckChecksums() } @@ -402,7 +403,7 @@ func uniffiCheckChecksums() { checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { return C.uniffi_iroh_streamplace_checksum_func_resign() }) - if checksum != 7588 { + if checksum != 33602 { // If this happens try cleaning and rebuilding your project panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_resign: UniFFI API checksum mismatch") } @@ -420,7 +421,7 @@ func uniffiCheckChecksums() { checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { return C.uniffi_iroh_streamplace_checksum_func_sign_with_ingredients() }) - if checksum != 34840 { + if checksum != 52190 { // If this happens try cleaning and rebuilding your project panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_sign_with_ingredients: UniFFI API checksum mismatch") } @@ -560,6 +561,15 @@ func uniffiCheckChecksums() { panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_gosigner_sign: UniFFI API checksum mismatch") } } + { + checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { + return C.uniffi_iroh_streamplace_checksum_method_manystreams_next() + }) + if checksum != 46617 { + // If this happens try cleaning and rebuilding your project + panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_manystreams_next: UniFFI API checksum mismatch") + } + } { checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { return C.uniffi_iroh_streamplace_checksum_method_node_add_tickets() @@ -1734,6 +1744,131 @@ func (c FfiConverterGoSigner) register() { C.uniffi_iroh_streamplace_fn_init_callback_vtable_gosigner(&UniffiVTableCallbackInterfaceGoSignerINSTANCE) } +type ManyStreams interface { + // Get the next stream from the many streams + Next() (*Stream, error) +} +type ManyStreamsImpl struct { + ffiObject FfiObject +} + +// Get the next stream from the many streams +func (_self *ManyStreamsImpl) Next() (*Stream, error) { + _pointer := _self.ffiObject.incrementPointer("ManyStreams") + defer _self.ffiObject.decrementPointer() + _uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI { + return GoRustBuffer{ + inner: C.uniffi_iroh_streamplace_fn_method_manystreams_next( + _pointer, _uniffiStatus), + } + }) + if _uniffiErr != nil { + var _uniffiDefaultValue *Stream + return _uniffiDefaultValue, _uniffiErr + } else { + return FfiConverterOptionalStreamINSTANCE.Lift(_uniffiRV), nil + } +} +func (object *ManyStreamsImpl) Destroy() { + runtime.SetFinalizer(object, nil) + object.ffiObject.destroy() +} + +type FfiConverterManyStreams struct { + handleMap *concurrentHandleMap[ManyStreams] +} + +var FfiConverterManyStreamsINSTANCE = FfiConverterManyStreams{ + handleMap: newConcurrentHandleMap[ManyStreams](), +} + +func (c FfiConverterManyStreams) Lift(pointer unsafe.Pointer) ManyStreams { + result := &ManyStreamsImpl{ + newFfiObject( + pointer, + func(pointer unsafe.Pointer, status *C.RustCallStatus) unsafe.Pointer { + return C.uniffi_iroh_streamplace_fn_clone_manystreams(pointer, status) + }, + func(pointer unsafe.Pointer, status *C.RustCallStatus) { + C.uniffi_iroh_streamplace_fn_free_manystreams(pointer, status) + }, + ), + } + runtime.SetFinalizer(result, (*ManyStreamsImpl).Destroy) + return result +} + +func (c FfiConverterManyStreams) Read(reader io.Reader) ManyStreams { + return c.Lift(unsafe.Pointer(uintptr(readUint64(reader)))) +} + +func (c FfiConverterManyStreams) Lower(value ManyStreams) unsafe.Pointer { + // TODO: this is bad - all synchronization from ObjectRuntime.go is discarded here, + // because the pointer will be decremented immediately after this function returns, + // and someone will be left holding onto a non-locked pointer. + pointer := unsafe.Pointer(uintptr(c.handleMap.insert(value))) + return pointer + +} + +func (c FfiConverterManyStreams) Write(writer io.Writer, value ManyStreams) { + writeUint64(writer, uint64(uintptr(c.Lower(value)))) +} + +type FfiDestroyerManyStreams struct{} + +func (_ FfiDestroyerManyStreams) Destroy(value ManyStreams) { + if val, ok := value.(*ManyStreamsImpl); ok { + val.Destroy() + } else { + panic("Expected *ManyStreamsImpl") + } +} + +//export iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsMethod0 +func iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsMethod0(uniffiHandle C.uint64_t, uniffiOutReturn *C.RustBuffer, callStatus *C.RustCallStatus) { + handle := uint64(uniffiHandle) + uniffiObj, ok := FfiConverterManyStreamsINSTANCE.handleMap.tryGet(handle) + if !ok { + panic(fmt.Errorf("no callback in handle map: %d", handle)) + } + + res, err := + uniffiObj.Next() + + if err != nil { + var actualError *SpError + if errors.As(err, &actualError) { + *callStatus = C.RustCallStatus{ + code: C.int8_t(uniffiCallbackResultError), + errorBuf: FfiConverterSpErrorINSTANCE.Lower(actualError), + } + } else { + *callStatus = C.RustCallStatus{ + code: C.int8_t(uniffiCallbackUnexpectedResultError), + } + } + return + } + + *uniffiOutReturn = FfiConverterOptionalStreamINSTANCE.Lower(res) +} + +var UniffiVTableCallbackInterfaceManyStreamsINSTANCE = C.UniffiVTableCallbackInterfaceManyStreams{ + next: (C.UniffiCallbackInterfaceManyStreamsMethod0)(C.iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsMethod0), + + uniffiFree: (C.UniffiCallbackInterfaceFree)(C.iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsFree), +} + +//export iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsFree +func iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsFree(handle C.uint64_t) { + FfiConverterManyStreamsINSTANCE.handleMap.remove(uint64(handle)) +} + +func (c FfiConverterManyStreams) register() { + C.uniffi_iroh_streamplace_fn_init_callback_vtable_manystreams(&UniffiVTableCallbackInterfaceManyStreamsINSTANCE) +} + // Iroh-streamplace node that can send, forward or receive stream segments. type NodeInterface interface { // Add tickets for remote peers @@ -4806,6 +4941,43 @@ func (_ FfiDestroyerOptionalBytes) Destroy(value *[]byte) { } } +type FfiConverterOptionalStream struct{} + +var FfiConverterOptionalStreamINSTANCE = FfiConverterOptionalStream{} + +func (c FfiConverterOptionalStream) Lift(rb RustBufferI) *Stream { + return LiftFromRustBuffer[*Stream](c, rb) +} + +func (_ FfiConverterOptionalStream) Read(reader io.Reader) *Stream { + if readInt8(reader) == 0 { + return nil + } + temp := FfiConverterStreamINSTANCE.Read(reader) + return &temp +} + +func (c FfiConverterOptionalStream) Lower(value *Stream) C.RustBuffer { + return LowerIntoRustBuffer[*Stream](c, value) +} + +func (_ FfiConverterOptionalStream) Write(writer io.Writer, value *Stream) { + if value == nil { + writeInt8(writer, 0) + } else { + writeInt8(writer, 1) + FfiConverterStreamINSTANCE.Write(writer, *value) + } +} + +type FfiDestroyerOptionalStream struct{} + +func (_ FfiDestroyerOptionalStream) Destroy(value *Stream) { + if value != nil { + FfiDestroyerStream{}.Destroy(*value) + } +} + type FfiConverterOptionalSubscribeItem struct{} var FfiConverterOptionalSubscribeItemINSTANCE = FfiConverterOptionalSubscribeItem{} @@ -5140,18 +5312,12 @@ func NodeIdFromTicket(ticketStr string) (*PublicKey, error) { } } -func Resign(unsignedSegData [][]byte, signedConcatData Stream, manifestList []string, certs [][]byte) ([][]byte, error) { - _uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI { - return GoRustBuffer{ - inner: C.uniffi_iroh_streamplace_fn_func_resign(FfiConverterSequenceBytesINSTANCE.Lower(unsignedSegData), FfiConverterStreamINSTANCE.Lower(signedConcatData), FfiConverterSequenceStringINSTANCE.Lower(manifestList), FfiConverterSequenceBytesINSTANCE.Lower(certs), _uniffiStatus), - } +func Resign(unsignedSegData ManyStreams, signedConcatData Stream, manifestList []string, certs [][]byte, output ManyStreams) error { + _, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) bool { + C.uniffi_iroh_streamplace_fn_func_resign(FfiConverterManyStreamsINSTANCE.Lower(unsignedSegData), FfiConverterStreamINSTANCE.Lower(signedConcatData), FfiConverterSequenceStringINSTANCE.Lower(manifestList), FfiConverterSequenceBytesINSTANCE.Lower(certs), FfiConverterManyStreamsINSTANCE.Lower(output), _uniffiStatus) + return false }) - if _uniffiErr != nil { - var _uniffiDefaultValue [][]byte - return _uniffiDefaultValue, _uniffiErr - } else { - return FfiConverterSequenceBytesINSTANCE.Lift(_uniffiRV), nil - } + return _uniffiErr.AsError() } func Sign(manifest string, data Stream, certs []byte, gosigner GoSigner) ([]byte, error) { @@ -5168,9 +5334,9 @@ func Sign(manifest string, data Stream, certs []byte, gosigner GoSigner) ([]byte } } -func SignWithIngredients(manifest string, data Stream, certs []byte, ingredients [][]byte, gosigner GoSigner, output Stream) error { +func SignWithIngredients(manifest string, data Stream, certs []byte, ingredients ManyStreams, gosigner GoSigner, output Stream) error { _, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) bool { - C.uniffi_iroh_streamplace_fn_func_sign_with_ingredients(FfiConverterStringINSTANCE.Lower(manifest), FfiConverterStreamINSTANCE.Lower(data), FfiConverterBytesINSTANCE.Lower(certs), FfiConverterSequenceBytesINSTANCE.Lower(ingredients), FfiConverterGoSignerINSTANCE.Lower(gosigner), FfiConverterStreamINSTANCE.Lower(output), _uniffiStatus) + C.uniffi_iroh_streamplace_fn_func_sign_with_ingredients(FfiConverterStringINSTANCE.Lower(manifest), FfiConverterStreamINSTANCE.Lower(data), FfiConverterBytesINSTANCE.Lower(certs), FfiConverterManyStreamsINSTANCE.Lower(ingredients), FfiConverterGoSignerINSTANCE.Lower(gosigner), FfiConverterStreamINSTANCE.Lower(output), _uniffiStatus) return false }) return _uniffiErr.AsError() diff --git a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h index e41afb4d4..4bdb73af7 100644 --- a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h +++ b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h @@ -405,6 +405,20 @@ static void call_UniffiCallbackInterfaceGoSignerMethod0( } +#endif +#ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_MANY_STREAMS_METHOD0 +#define UNIFFI_FFIDEF_CALLBACK_INTERFACE_MANY_STREAMS_METHOD0 +typedef void (*UniffiCallbackInterfaceManyStreamsMethod0)(uint64_t uniffi_handle, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); + +// Making function static works arround: +// https://github.com/golang/go/issues/11263 +static void call_UniffiCallbackInterfaceManyStreamsMethod0( + UniffiCallbackInterfaceManyStreamsMethod0 cb, uint64_t uniffi_handle, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ) +{ + return cb(uniffi_handle, uniffi_out_return, callStatus ); +} + + #endif #ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_STREAM_METHOD0 #define UNIFFI_FFIDEF_CALLBACK_INTERFACE_STREAM_METHOD0 @@ -463,6 +477,14 @@ typedef struct UniffiVTableCallbackInterfaceGoSigner { UniffiCallbackInterfaceFree uniffiFree; } UniffiVTableCallbackInterfaceGoSigner; +#endif +#ifndef UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_MANY_STREAMS +#define UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_MANY_STREAMS +typedef struct UniffiVTableCallbackInterfaceManyStreams { + UniffiCallbackInterfaceManyStreamsMethod0 next; + UniffiCallbackInterfaceFree uniffiFree; +} UniffiVTableCallbackInterfaceManyStreams; + #endif #ifndef UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_STREAM #define UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_STREAM @@ -600,6 +622,26 @@ void uniffi_iroh_streamplace_fn_init_callback_vtable_gosigner(UniffiVTableCallba RustBuffer uniffi_iroh_streamplace_fn_method_gosigner_sign(void* ptr, RustBuffer data, RustCallStatus *out_status ); #endif +#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_MANYSTREAMS +#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_MANYSTREAMS +void* uniffi_iroh_streamplace_fn_clone_manystreams(void* ptr, RustCallStatus *out_status +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_MANYSTREAMS +#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_MANYSTREAMS +void uniffi_iroh_streamplace_fn_free_manystreams(void* ptr, RustCallStatus *out_status +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_INIT_CALLBACK_VTABLE_MANYSTREAMS +#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_INIT_CALLBACK_VTABLE_MANYSTREAMS +void uniffi_iroh_streamplace_fn_init_callback_vtable_manystreams(UniffiVTableCallbackInterfaceManyStreams* vtable +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_MANYSTREAMS_NEXT +#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_MANYSTREAMS_NEXT +RustBuffer uniffi_iroh_streamplace_fn_method_manystreams_next(void* ptr, RustCallStatus *out_status +); +#endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_NODE #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_NODE void* uniffi_iroh_streamplace_fn_clone_node(void* ptr, RustCallStatus *out_status @@ -843,7 +885,7 @@ void* uniffi_iroh_streamplace_fn_func_node_id_from_ticket(RustBuffer ticket_str, #endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_RESIGN #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_RESIGN -RustBuffer uniffi_iroh_streamplace_fn_func_resign(RustBuffer unsigned_seg_data, void* signed_concat_data, RustBuffer manifest_list, RustBuffer certs, RustCallStatus *out_status +void uniffi_iroh_streamplace_fn_func_resign(void* unsigned_seg_data, void* signed_concat_data, RustBuffer manifest_list, RustBuffer certs, void* output, RustCallStatus *out_status ); #endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN @@ -853,7 +895,7 @@ RustBuffer uniffi_iroh_streamplace_fn_func_sign(RustBuffer manifest, void* data, #endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN_WITH_INGREDIENTS #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN_WITH_INGREDIENTS -void uniffi_iroh_streamplace_fn_func_sign_with_ingredients(RustBuffer manifest, void* data, RustBuffer certs, RustBuffer ingredients, void* gosigner, void* output, RustCallStatus *out_status +void uniffi_iroh_streamplace_fn_func_sign_with_ingredients(RustBuffer manifest, void* data, RustBuffer certs, void* ingredients, void* gosigner, void* output, RustCallStatus *out_status ); #endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SUBSCRIBE_ITEM_DEBUG @@ -1277,6 +1319,12 @@ uint16_t uniffi_iroh_streamplace_checksum_method_filter_timestamps(void #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_GOSIGNER_SIGN uint16_t uniffi_iroh_streamplace_checksum_method_gosigner_sign(void +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_MANYSTREAMS_NEXT +#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_MANYSTREAMS_NEXT +uint16_t uniffi_iroh_streamplace_checksum_method_manystreams_next(void + ); #endif #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_NODE_ADD_TICKETS @@ -1464,6 +1512,8 @@ uint32_t ffi_iroh_streamplace_uniffi_contract_version(void void iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerFree(uint64_t handle); void iroh_streamplace_cgo_dispatchCallbackInterfaceGoSignerMethod0(uint64_t uniffi_handle, RustBuffer data, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); void iroh_streamplace_cgo_dispatchCallbackInterfaceGoSignerFree(uint64_t handle); + void iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsMethod0(uint64_t uniffi_handle, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); + void iroh_streamplace_cgo_dispatchCallbackInterfaceManyStreamsFree(uint64_t handle); void iroh_streamplace_cgo_dispatchCallbackInterfaceStreamMethod0(uint64_t uniffi_handle, uint64_t length, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); void iroh_streamplace_cgo_dispatchCallbackInterfaceStreamMethod1(uint64_t uniffi_handle, int64_t pos, uint64_t mode, uint64_t* uniffi_out_return, RustCallStatus* callStatus ); void iroh_streamplace_cgo_dispatchCallbackInterfaceStreamMethod2(uint64_t uniffi_handle, RustBuffer data, uint64_t* uniffi_out_return, RustCallStatus* callStatus ); diff --git a/pkg/media/media_signer.go b/pkg/media/media_signer.go index 3396fdeef..706eef174 100644 --- a/pkg/media/media_signer.go +++ b/pkg/media/media_signer.go @@ -228,8 +228,12 @@ func (ms *MediaSignerLocal) SignConcatMP4(ctx context.Context, input io.ReadSeek rustCallbackSigner := &RustCallbackSigner{ Signer: ms.Signer, } + many := c2patypes.NewManyStreams() + for _, ingredient := range ingredients { + many.AddStream(aqio.NewReadWriteSeeker(ingredient)) + } rws := aqio.NewReadWriteSeeker([]byte{}) - err = iroh_streamplace.SignWithIngredients(string(manifestBs), c2patypes.NewReader(input), ms.Cert, ingredients, rustCallbackSigner, c2patypes.NewWriter(rws)) + err = iroh_streamplace.SignWithIngredients(string(manifestBs), c2patypes.NewReader(input), ms.Cert, many, rustCallbackSigner, c2patypes.NewWriter(rws)) if err != nil { return nil, err } diff --git a/pkg/media/segment_split.go b/pkg/media/segment_split.go index 35aa12aa6..026065b3e 100644 --- a/pkg/media/segment_split.go +++ b/pkg/media/segment_split.go @@ -69,17 +69,28 @@ func SplitSegments(ctx context.Context, input []byte) ([]SplitSegment, error) { if err != nil { return nil, fmt.Errorf("failed to segment file: %w", err) } - resignedSegs, err := iroh_streamplace.Resign(unsignedSegs, c2patypes.NewReader(aqio.NewReadWriteSeeker(input)), manifestStrs, certList) + inputStreams := c2patypes.NewManyStreams() + outputStreams := c2patypes.NewManyStreams() + for _, unsignedSeg := range unsignedSegs { + inputStreams.AddStream(aqio.NewReadWriteSeeker(unsignedSeg)) + outputStreams.AddStream(aqio.NewReadWriteSeeker([]byte{})) + } + err = iroh_streamplace.Resign(inputStreams, c2patypes.NewReader(aqio.NewReadWriteSeeker(input)), manifestStrs, certList, outputStreams) if err != nil { return nil, fmt.Errorf("failed to resign segments: %w", err) } splitSegments := []SplitSegment{} - for i, resignedSeg := range resignedSegs { + for i, resignedSeg := range outputStreams.Streams { + aqrws := resignedSeg.(*aqio.ReadWriteSeeker) + data, err := aqrws.Bytes() + if err != nil { + return nil, fmt.Errorf("failed to read resigned segment: %w", err) + } meta := manifestList[i].SegmentMetadata fname := fmt.Sprintf("%s.mp4", meta.StartTime.FileSafeString()) splitSegments = append(splitSegments, SplitSegment{ Filename: fname, - Data: resignedSeg, + Data: data, }) } return splitSegments, nil diff --git a/rust/iroh-streamplace/src/c2pa.rs b/rust/iroh-streamplace/src/c2pa.rs index 9437505db..90b950cd5 100644 --- a/rust/iroh-streamplace/src/c2pa.rs +++ b/rust/iroh-streamplace/src/c2pa.rs @@ -10,11 +10,10 @@ use c2pa::jumbf_io; use c2pa::status_tracker::StatusTracker; use c2pa::store::Store; +use crate::streams::ManyStreams; use crate::streams::Stream; use crate::streams::StreamAdapter; - use serde_json; -use tracing::info; use crate::error::SPError; #[uniffi::export] @@ -164,11 +163,12 @@ pub fn get_manifests(data: &dyn Stream) -> Result { #[uniffi::export] pub fn resign( - unsigned_seg_data: Vec>, + unsigned_seg_data: &dyn ManyStreams, signed_concat_data: &dyn Stream, manifest_list: Vec, certs: Vec>, -) -> Result>, SPError> { + output: &dyn ManyStreams, +) -> Result<(), SPError> { let mut validation_log = StatusTracker::default(); let combined_store = Store::from_stream( @@ -179,16 +179,31 @@ pub fn resign( ) .map_err(|e| SPError::C2paError(format!("from_stream failed: {}", e)))?; - // let seg_claim = combined_store - // .get_claim(unsigned_seg_label.as_str()) - // .ok_or(SPError::C2paError(format!( - // "Segment claim not found: {}", - // unsigned_seg_label - // )))?; - let mut outputs = Vec::>::new(); - for (i, manifest_id) in manifest_list.iter().enumerate() { - let mut output = Vec::::new(); + let unsigned_seg_data_stream = match unsigned_seg_data.next() { + Ok(Some(stream)) => stream, + Ok(None) => return Err(SPError::C2paError("Input stream not found".to_string())), + Err(e) => { + return Err(SPError::C2paError(format!( + "Error getting input stream: {}", + e + ))); + } + }; + let output_stream = match output.next() { + Ok(Some(stream)) => stream, + Ok(None) => return Err(SPError::C2paError("Output stream not found".to_string())), + Err(e) => { + return Err(SPError::C2paError(format!( + "Error getting output stream: {}", + e + ))); + } + }; + let mut input_cursor = StreamAdapter::from(&*unsigned_seg_data_stream); + let mut output_cursor = StreamAdapter::from(&*output_stream); + + // let mut output = Vec::::new(); let seg_claim = combined_store .get_claim(manifest_id.as_str()) @@ -209,9 +224,6 @@ pub fn resign( .commit_claim(seg_claim.clone()) .map_err(|e| SPError::C2paError(format!("commit_claim failed: {}", e)))?; - let mut output_cursor = Cursor::new(&mut output); - let mut input_cursor = Cursor::new(unsigned_seg_data[i].clone()); - let jumbf_bytes = seg_store .to_jumbf(&callback_signer) .map_err(|e| SPError::C2paError(format!("to_jumbf failed: {}", e)))?; @@ -223,10 +235,9 @@ pub fn resign( &jumbf_bytes, ) .map_err(|e| SPError::C2paError(format!("save_jumbf_to_stream failed: {}", e)))?; - outputs.push(output); } - Ok(outputs) + Ok(()) } #[uniffi::export] @@ -234,7 +245,7 @@ pub fn sign_with_ingredients( manifest: String, data: &dyn Stream, certs: Vec, - ingredients: Vec>, + ingredients: &dyn ManyStreams, gosigner: Arc, output: &dyn Stream, ) -> Result<(), SPError> { @@ -250,8 +261,8 @@ pub fn sign_with_ingredients( ); let mut builder = Builder::from_json(&manifest).map_err(|e| SPError::C2paError(e.to_string()))?; - for (i, ingredient) in ingredients.into_iter().enumerate() { - let mut cursor = Cursor::new(ingredient); + while let Ok(Some(ingredient)) = ingredients.next() { + let mut cursor = StreamAdapter::from(&*ingredient); let ingredient = Ingredient::from_stream("video/mp4", &mut cursor) .map_err(|e| SPError::C2paError(e.to_string()))?; builder.add_ingredient(ingredient); diff --git a/rust/iroh-streamplace/src/streams.rs b/rust/iroh-streamplace/src/streams.rs index e8e1d4135..d3dd536f5 100644 --- a/rust/iroh-streamplace/src/streams.rs +++ b/rust/iroh-streamplace/src/streams.rs @@ -124,6 +124,12 @@ impl<'a> Write for StreamAdapter<'a> { } } +#[uniffi::export(with_foreign)] +pub trait ManyStreams: Send + Sync { + /// Get the next stream from the many streams + fn next(&self) -> Result>, SPError>; +} + #[cfg(test)] mod tests { use super::*;