+1 -1

pkg/api/api.go

··· 497 497 498 498 func (a *StreamplaceAPI) HandleSegment(ctx context.Context) http.HandlerFunc { 499 499 return func(w http.ResponseWriter, req *http.Request) { 500 500 - err := a.MediaManager.ValidateMP4(ctx, req.Body) 500 500 + err := a.MediaManager.ValidateMP4(ctx, req.Body, false) 501 501 if err != nil { 502 502 apierrors.WriteHTTPBadRequest(w, "could not ingest segment", err) 503 503 return

+3 -6

pkg/cmd/streamplace.go

··· 32 32 "stream.place/streamplace/pkg/log" 33 33 "stream.place/streamplace/pkg/media" 34 34 "stream.place/streamplace/pkg/notifications" 35 35 - "stream.place/streamplace/pkg/replication" 36 36 - "stream.place/streamplace/pkg/replication/boring" 37 35 "stream.place/streamplace/pkg/replication/iroh_replicator" 38 36 "stream.place/streamplace/pkg/rtmps" 39 37 v0 "stream.place/streamplace/pkg/schema/v0" ··· 307 305 log.Log(ctx, "successfully initialized hardware signer", "address", addr) 308 306 signer = hwsigner 309 307 } 310 310 - var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers} 311 308 312 309 mod, err := model.MakeDB(cli.DataFilePath([]string{"index"})) 313 310 if err != nil { ··· 363 360 return fmt.Errorf("failed to migrate: %w", err) 364 361 } 365 362 366 366 - mm, err := media.MakeMediaManager(ctx, &cli, signer, rep, mod, b, atsync) 363 363 + mm, err := media.MakeMediaManager(ctx, &cli, signer, mod, b, atsync) 367 364 if err != nil { 368 365 return err 369 366 } ··· 403 400 return err 404 401 } 405 402 secret := buf.Bytes() 406 406 - swarm, err := iroh_replicator.StartKV(ctx, cli.Tickets, secret) 403 403 + swarm, err := iroh_replicator.NewSwarm(ctx, cli.Tickets, secret, mm) 407 404 if err != nil { 408 405 return err 409 406 } ··· 419 416 DownstreamJWK: cli.AccessJWK, 420 417 ClientMetadata: clientMetadata, 421 418 }) 422 422 - d := director.NewDirector(mm, mod, &cli, b, op, state, swarm) 419 419 + d := director.NewDirector(mm, mod, &cli, b, op, state) 423 420 a, err := api.MakeStreamplaceAPI(&cli, mod, state, eip712signer, noter, mm, ms, b, atsync, d, op) 424 421 if err != nil { 425 422 return err

+1 -1

pkg/config/config.go

··· 167 167 fs.StringVar(&cli.PublicHost, "public-host", "", "public host for this streamplace node (excluding https:// e.g. stream.place)") 168 168 fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation") 169 169 fs.BoolVar(&cli.SmearAudio, "smear-audio", false, "enable audio smearing to create 'perfect' segment timestamps") 170 170 - fs.BoolVar(&cli.ExternalSigning, "external-signing", true, "enable external signing via exec (prevents potential memory leak)") 170 170 + fs.BoolVar(&cli.ExternalSigning, "external-signing", false, "enable external signing via exec (prevents potential memory leak)") 171 171 fs.StringVar(&cli.TracingEndpoint, "tracing-endpoint", "", "gRPC endpoint to send traces to") 172 172 fs.IntVar(&cli.RateLimitPerSecond, "rate-limit-per-second", 0, "rate limit for requests per second per ip") 173 173 fs.IntVar(&cli.RateLimitBurst, "rate-limit-burst", 0, "rate limit burst for requests per ip")

+2 -32

pkg/director/director.go

··· 2 2 3 3 import ( 4 4 "context" 5 5 - "encoding/json" 6 5 "fmt" 7 6 "sync" 8 7 9 9 - "github.com/bluesky-social/indigo/util" 10 8 "github.com/streamplace/oatproxy/pkg/oatproxy" 11 9 "golang.org/x/sync/errgroup" 12 10 "stream.place/streamplace/pkg/bus" ··· 14 12 "stream.place/streamplace/pkg/log" 15 13 "stream.place/streamplace/pkg/media" 16 14 "stream.place/streamplace/pkg/model" 17 17 - "stream.place/streamplace/pkg/replication/iroh_replicator" 18 15 "stream.place/streamplace/pkg/statedb" 19 16 ) 20 17 ··· 33 30 streamSessionsMu sync.Mutex 34 31 op *oatproxy.OATProxy 35 32 statefulDB *statedb.StatefulDB 36 36 - swarm *iroh_replicator.SwarmKV 37 33 } 38 34 39 39 - func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB, swarm *iroh_replicator.SwarmKV) *Director { 35 35 + func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB) *Director { 40 36 return &Director{ 41 37 mm: mm, 42 38 mod: mod, ··· 46 42 streamSessionsMu: sync.Mutex{}, 47 43 op: op, 48 44 statefulDB: statefulDB, 49 49 - swarm: swarm, 50 45 } 51 46 } 52 47 53 48 func (d *Director) Start(ctx context.Context) error { 54 54 - nodeId, err := d.swarm.Node.NodeId() 55 55 - if err != nil { 56 56 - return fmt.Errorf("failed to get node id: %w", err) 57 57 - } 58 58 - 59 49 newSeg := d.mm.NewSegment() 60 50 ctx, cancel := context.WithCancel(ctx) 61 51 defer cancel() ··· 96 86 }) 97 87 } 98 88 d.streamSessionsMu.Unlock() 99 99 - go func() { 100 100 - originInfo := iroh_replicator.OriginInfo{ 101 101 - NodeID: nodeId.String(), 102 102 - Time: not.Segment.StartTime.Format(util.ISO8601), 103 103 - } 104 104 - bs, err := json.Marshal(originInfo) 105 105 - if err != nil { 106 106 - log.Error(ctx, "could not marshal origin info", "error", err) 107 107 - return 108 108 - } 109 109 - err = d.swarm.Put(ctx, not.Segment.RepoDID, bs) 110 110 - if err != nil { 111 111 - log.Error(ctx, "could not put segment to swarm", "error", err) 112 112 - return 113 113 - } 114 114 - err = d.swarm.Node.SendSegment(not.Segment.RepoDID, not.Data) 115 115 - if err != nil { 116 116 - log.Error(ctx, "could not send segment to swarm", "error", err) 117 117 - return 118 118 - } 119 119 - }() 89 89 + 120 90 err := ss.NewSegment(ctx, not) 121 91 if err != nil { 122 92 log.Error(ctx, "could not add segment to stream session", "error", err)

-681

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go

··· 337 337 func init() { 338 338 339 339 FfiConverterDataHandlerINSTANCE.register() 340 340 - FfiConverterDataHandlerOldINSTANCE.register() 341 340 FfiConverterGoSignerINSTANCE.register() 342 341 uniffiCheckChecksums() 343 342 } ··· 391 390 } 392 391 { 393 392 checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 394 394 - return C.uniffi_iroh_streamplace_checksum_method_datahandlerold_handle_data() 395 395 - }) 396 396 - if checksum != 20343 { 397 397 - // If this happens try cleaning and rebuilding your project 398 398 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_datahandlerold_handle_data: UniFFI API checksum mismatch") 399 399 - } 400 400 - } 401 401 - { 402 402 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 403 393 return C.uniffi_iroh_streamplace_checksum_method_db_iter_with_opts() 404 394 }) 405 395 if checksum != 3815 { ··· 432 422 if checksum != 5247 { 433 423 // If this happens try cleaning and rebuilding your project 434 424 panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_db_write: UniFFI API checksum mismatch") 435 435 - } 436 436 - } 437 437 - { 438 438 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 439 439 - return C.uniffi_iroh_streamplace_checksum_method_endpoint_node_addr() 440 440 - }) 441 441 - if checksum != 17254 { 442 442 - // If this happens try cleaning and rebuilding your project 443 443 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_endpoint_node_addr: UniFFI API checksum mismatch") 444 425 } 445 426 } 446 427 { ··· 652 633 } 653 634 { 654 635 checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 655 655 - return C.uniffi_iroh_streamplace_checksum_method_receiver_node_addr() 656 656 - }) 657 657 - if checksum != 10730 { 658 658 - // If this happens try cleaning and rebuilding your project 659 659 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_receiver_node_addr: UniFFI API checksum mismatch") 660 660 - } 661 661 - } 662 662 - { 663 663 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 664 664 - return C.uniffi_iroh_streamplace_checksum_method_receiver_subscribe() 665 665 - }) 666 666 - if checksum != 24145 { 667 667 - // If this happens try cleaning and rebuilding your project 668 668 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_receiver_subscribe: UniFFI API checksum mismatch") 669 669 - } 670 670 - } 671 671 - { 672 672 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 673 673 - return C.uniffi_iroh_streamplace_checksum_method_receiver_unsubscribe() 674 674 - }) 675 675 - if checksum != 21760 { 676 676 - // If this happens try cleaning and rebuilding your project 677 677 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_receiver_unsubscribe: UniFFI API checksum mismatch") 678 678 - } 679 679 - } 680 680 - { 681 681 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 682 682 - return C.uniffi_iroh_streamplace_checksum_method_sender_node_addr() 683 683 - }) 684 684 - if checksum != 38541 { 685 685 - // If this happens try cleaning and rebuilding your project 686 686 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_sender_node_addr: UniFFI API checksum mismatch") 687 687 - } 688 688 - } 689 689 - { 690 690 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 691 691 - return C.uniffi_iroh_streamplace_checksum_method_sender_send() 692 692 - }) 693 693 - if checksum != 23930 { 694 694 - // If this happens try cleaning and rebuilding your project 695 695 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_sender_send: UniFFI API checksum mismatch") 696 696 - } 697 697 - } 698 698 - { 699 699 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 700 636 return C.uniffi_iroh_streamplace_checksum_method_subscriberesponse_next_raw() 701 637 }) 702 638 if checksum != 55650 { ··· 711 647 if checksum != 50543 { 712 648 // If this happens try cleaning and rebuilding your project 713 649 panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_method_writescope_put: UniFFI API checksum mismatch") 714 714 - } 715 715 - } 716 716 - { 717 717 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 718 718 - return C.uniffi_iroh_streamplace_checksum_constructor_endpoint_new() 719 719 - }) 720 720 - if checksum != 60672 { 721 721 - // If this happens try cleaning and rebuilding your project 722 722 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_constructor_endpoint_new: UniFFI API checksum mismatch") 723 650 } 724 651 } 725 652 { ··· 785 712 panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_constructor_publickey_from_string: UniFFI API checksum mismatch") 786 713 } 787 714 } 788 788 - { 789 789 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 790 790 - return C.uniffi_iroh_streamplace_checksum_constructor_receiver_new() 791 791 - }) 792 792 - if checksum != 18072 { 793 793 - // If this happens try cleaning and rebuilding your project 794 794 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_constructor_receiver_new: UniFFI API checksum mismatch") 795 795 - } 796 796 - } 797 797 - { 798 798 - checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { 799 799 - return C.uniffi_iroh_streamplace_checksum_constructor_sender_new() 800 800 - }) 801 801 - if checksum != 56457 { 802 802 - // If this happens try cleaning and rebuilding your project 803 803 - panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_constructor_sender_new: UniFFI API checksum mismatch") 804 804 - } 805 805 - } 806 715 } 807 716 808 717 type FfiConverterUint64 struct{} ··· 1251 1160 C.uniffi_iroh_streamplace_fn_init_callback_vtable_datahandler(&UniffiVTableCallbackInterfaceDataHandlerINSTANCE) 1252 1161 } 1253 1162 1254 1254 - type DataHandlerOld interface { 1255 1255 - HandleData(topic string, data []byte) 1256 1256 - } 1257 1257 - type DataHandlerOldImpl struct { 1258 1258 - ffiObject FfiObject 1259 1259 - } 1260 1260 - 1261 1261 - func (_self *DataHandlerOldImpl) HandleData(topic string, data []byte) { 1262 1262 - _pointer := _self.ffiObject.incrementPointer("DataHandlerOld") 1263 1263 - defer _self.ffiObject.decrementPointer() 1264 1264 - uniffiRustCallAsync[error]( 1265 1265 - nil, 1266 1266 - // completeFn 1267 1267 - func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 1268 1268 - C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 1269 1269 - return struct{}{} 1270 1270 - }, 1271 1271 - // liftFn 1272 1272 - func(_ struct{}) struct{} { return struct{}{} }, 1273 1273 - C.uniffi_iroh_streamplace_fn_method_datahandlerold_handle_data( 1274 1274 - _pointer, FfiConverterStringINSTANCE.Lower(topic), FfiConverterBytesINSTANCE.Lower(data)), 1275 1275 - // pollFn 1276 1276 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1277 1277 - C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 1278 1278 - }, 1279 1279 - // freeFn 1280 1280 - func(handle C.uint64_t) { 1281 1281 - C.ffi_iroh_streamplace_rust_future_free_void(handle) 1282 1282 - }, 1283 1283 - ) 1284 1284 - 1285 1285 - } 1286 1286 - func (object *DataHandlerOldImpl) Destroy() { 1287 1287 - runtime.SetFinalizer(object, nil) 1288 1288 - object.ffiObject.destroy() 1289 1289 - } 1290 1290 - 1291 1291 - type FfiConverterDataHandlerOld struct { 1292 1292 - handleMap *concurrentHandleMap[DataHandlerOld] 1293 1293 - } 1294 1294 - 1295 1295 - var FfiConverterDataHandlerOldINSTANCE = FfiConverterDataHandlerOld{ 1296 1296 - handleMap: newConcurrentHandleMap[DataHandlerOld](), 1297 1297 - } 1298 1298 - 1299 1299 - func (c FfiConverterDataHandlerOld) Lift(pointer unsafe.Pointer) DataHandlerOld { 1300 1300 - result := &DataHandlerOldImpl{ 1301 1301 - newFfiObject( 1302 1302 - pointer, 1303 1303 - func(pointer unsafe.Pointer, status *C.RustCallStatus) unsafe.Pointer { 1304 1304 - return C.uniffi_iroh_streamplace_fn_clone_datahandlerold(pointer, status) 1305 1305 - }, 1306 1306 - func(pointer unsafe.Pointer, status *C.RustCallStatus) { 1307 1307 - C.uniffi_iroh_streamplace_fn_free_datahandlerold(pointer, status) 1308 1308 - }, 1309 1309 - ), 1310 1310 - } 1311 1311 - runtime.SetFinalizer(result, (*DataHandlerOldImpl).Destroy) 1312 1312 - return result 1313 1313 - } 1314 1314 - 1315 1315 - func (c FfiConverterDataHandlerOld) Read(reader io.Reader) DataHandlerOld { 1316 1316 - return c.Lift(unsafe.Pointer(uintptr(readUint64(reader)))) 1317 1317 - } 1318 1318 - 1319 1319 - func (c FfiConverterDataHandlerOld) Lower(value DataHandlerOld) unsafe.Pointer { 1320 1320 - // TODO: this is bad - all synchronization from ObjectRuntime.go is discarded here, 1321 1321 - // because the pointer will be decremented immediately after this function returns, 1322 1322 - // and someone will be left holding onto a non-locked pointer. 1323 1323 - pointer := unsafe.Pointer(uintptr(c.handleMap.insert(value))) 1324 1324 - return pointer 1325 1325 - 1326 1326 - } 1327 1327 - 1328 1328 - func (c FfiConverterDataHandlerOld) Write(writer io.Writer, value DataHandlerOld) { 1329 1329 - writeUint64(writer, uint64(uintptr(c.Lower(value)))) 1330 1330 - } 1331 1331 - 1332 1332 - type FfiDestroyerDataHandlerOld struct{} 1333 1333 - 1334 1334 - func (_ FfiDestroyerDataHandlerOld) Destroy(value DataHandlerOld) { 1335 1335 - if val, ok := value.(*DataHandlerOldImpl); ok { 1336 1336 - val.Destroy() 1337 1337 - } else { 1338 1338 - panic("Expected *DataHandlerOldImpl") 1339 1339 - } 1340 1340 - } 1341 1341 - 1342 1342 - //export iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldMethod0 1343 1343 - func iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldMethod0(uniffiHandle C.uint64_t, topic C.RustBuffer, data C.RustBuffer, uniffiFutureCallback C.UniffiForeignFutureCompleteVoid, uniffiCallbackData C.uint64_t, uniffiOutReturn *C.UniffiForeignFuture) { 1344 1344 - handle := uint64(uniffiHandle) 1345 1345 - uniffiObj, ok := FfiConverterDataHandlerOldINSTANCE.handleMap.tryGet(handle) 1346 1346 - if !ok { 1347 1347 - panic(fmt.Errorf("no callback in handle map: %d", handle)) 1348 1348 - } 1349 1349 - 1350 1350 - result := make(chan C.UniffiForeignFutureStructVoid, 1) 1351 1351 - cancel := make(chan struct{}, 1) 1352 1352 - guardHandle := cgo.NewHandle(cancel) 1353 1353 - *uniffiOutReturn = C.UniffiForeignFuture{ 1354 1354 - handle: C.uint64_t(guardHandle), 1355 1355 - free: C.UniffiForeignFutureFree(C.iroh_streamplace_uniffiFreeGorutine), 1356 1356 - } 1357 1357 - 1358 1358 - // Wait for compleation or cancel 1359 1359 - go func() { 1360 1360 - select { 1361 1361 - case <-cancel: 1362 1362 - case res := <-result: 1363 1363 - C.call_UniffiForeignFutureCompleteVoid(uniffiFutureCallback, uniffiCallbackData, res) 1364 1364 - } 1365 1365 - }() 1366 1366 - 1367 1367 - // Eval callback asynchroniously 1368 1368 - go func() { 1369 1369 - asyncResult := &C.UniffiForeignFutureStructVoid{} 1370 1370 - defer func() { 1371 1371 - result <- *asyncResult 1372 1372 - }() 1373 1373 - 1374 1374 - uniffiObj.HandleData( 1375 1375 - FfiConverterStringINSTANCE.Lift(GoRustBuffer{ 1376 1376 - inner: topic, 1377 1377 - }), 1378 1378 - FfiConverterBytesINSTANCE.Lift(GoRustBuffer{ 1379 1379 - inner: data, 1380 1380 - }), 1381 1381 - ) 1382 1382 - 1383 1383 - }() 1384 1384 - } 1385 1385 - 1386 1386 - var UniffiVTableCallbackInterfaceDataHandlerOldINSTANCE = C.UniffiVTableCallbackInterfaceDataHandlerOld{ 1387 1387 - handleData: (C.UniffiCallbackInterfaceDataHandlerOldMethod0)(C.iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldMethod0), 1388 1388 - 1389 1389 - uniffiFree: (C.UniffiCallbackInterfaceFree)(C.iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldFree), 1390 1390 - } 1391 1391 - 1392 1392 - //export iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldFree 1393 1393 - func iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldFree(handle C.uint64_t) { 1394 1394 - FfiConverterDataHandlerOldINSTANCE.handleMap.remove(uint64(handle)) 1395 1395 - } 1396 1396 - 1397 1397 - func (c FfiConverterDataHandlerOld) register() { 1398 1398 - C.uniffi_iroh_streamplace_fn_init_callback_vtable_datahandlerold(&UniffiVTableCallbackInterfaceDataHandlerOldINSTANCE) 1399 1399 - } 1400 1400 - 1401 1163 // Iroh-streamplace specific metadata database. 1402 1164 type DbInterface interface { 1403 1165 IterWithOpts(filter *Filter) ([]Entry, error) ··· 1529 1291 value.Destroy() 1530 1292 } 1531 1293 1532 1532 - type EndpointInterface interface { 1533 1533 - NodeAddr() *NodeAddr 1534 1534 - } 1535 1535 - type Endpoint struct { 1536 1536 - ffiObject FfiObject 1537 1537 - } 1538 1538 - 1539 1539 - // Create a new endpoint. 1540 1540 - func NewEndpoint() (*Endpoint, error) { 1541 1541 - res, err := uniffiRustCallAsync[Error]( 1542 1542 - FfiConverterErrorINSTANCE, 1543 1543 - // completeFn 1544 1544 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 1545 1545 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 1546 1546 - return res 1547 1547 - }, 1548 1548 - // liftFn 1549 1549 - func(ffi unsafe.Pointer) *Endpoint { 1550 1550 - return FfiConverterEndpointINSTANCE.Lift(ffi) 1551 1551 - }, 1552 1552 - C.uniffi_iroh_streamplace_fn_constructor_endpoint_new(), 1553 1553 - // pollFn 1554 1554 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1555 1555 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 1556 1556 - }, 1557 1557 - // freeFn 1558 1558 - func(handle C.uint64_t) { 1559 1559 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 1560 1560 - }, 1561 1561 - ) 1562 1562 - 1563 1563 - if err == nil { 1564 1564 - return res, nil 1565 1565 - } 1566 1566 - 1567 1567 - return res, err 1568 1568 - } 1569 1569 - 1570 1570 - func (_self *Endpoint) NodeAddr() *NodeAddr { 1571 1571 - _pointer := _self.ffiObject.incrementPointer("*Endpoint") 1572 1572 - defer _self.ffiObject.decrementPointer() 1573 1573 - res, _ := uniffiRustCallAsync[error]( 1574 1574 - nil, 1575 1575 - // completeFn 1576 1576 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 1577 1577 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 1578 1578 - return res 1579 1579 - }, 1580 1580 - // liftFn 1581 1581 - func(ffi unsafe.Pointer) *NodeAddr { 1582 1582 - return FfiConverterNodeAddrINSTANCE.Lift(ffi) 1583 1583 - }, 1584 1584 - C.uniffi_iroh_streamplace_fn_method_endpoint_node_addr( 1585 1585 - _pointer), 1586 1586 - // pollFn 1587 1587 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 1588 1588 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 1589 1589 - }, 1590 1590 - // freeFn 1591 1591 - func(handle C.uint64_t) { 1592 1592 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 1593 1593 - }, 1594 1594 - ) 1595 1595 - 1596 1596 - return res 1597 1597 - } 1598 1598 - func (object *Endpoint) Destroy() { 1599 1599 - runtime.SetFinalizer(object, nil) 1600 1600 - object.ffiObject.destroy() 1601 1601 - } 1602 1602 - 1603 1603 - type FfiConverterEndpoint struct{} 1604 1604 - 1605 1605 - var FfiConverterEndpointINSTANCE = FfiConverterEndpoint{} 1606 1606 - 1607 1607 - func (c FfiConverterEndpoint) Lift(pointer unsafe.Pointer) *Endpoint { 1608 1608 - result := &Endpoint{ 1609 1609 - newFfiObject( 1610 1610 - pointer, 1611 1611 - func(pointer unsafe.Pointer, status *C.RustCallStatus) unsafe.Pointer { 1612 1612 - return C.uniffi_iroh_streamplace_fn_clone_endpoint(pointer, status) 1613 1613 - }, 1614 1614 - func(pointer unsafe.Pointer, status *C.RustCallStatus) { 1615 1615 - C.uniffi_iroh_streamplace_fn_free_endpoint(pointer, status) 1616 1616 - }, 1617 1617 - ), 1618 1618 - } 1619 1619 - runtime.SetFinalizer(result, (*Endpoint).Destroy) 1620 1620 - return result 1621 1621 - } 1622 1622 - 1623 1623 - func (c FfiConverterEndpoint) Read(reader io.Reader) *Endpoint { 1624 1624 - return c.Lift(unsafe.Pointer(uintptr(readUint64(reader)))) 1625 1625 - } 1626 1626 - 1627 1627 - func (c FfiConverterEndpoint) Lower(value *Endpoint) unsafe.Pointer { 1628 1628 - // TODO: this is bad - all synchronization from ObjectRuntime.go is discarded here, 1629 1629 - // because the pointer will be decremented immediately after this function returns, 1630 1630 - // and someone will be left holding onto a non-locked pointer. 1631 1631 - pointer := value.ffiObject.incrementPointer("*Endpoint") 1632 1632 - defer value.ffiObject.decrementPointer() 1633 1633 - return pointer 1634 1634 - 1635 1635 - } 1636 1636 - 1637 1637 - func (c FfiConverterEndpoint) Write(writer io.Writer, value *Endpoint) { 1638 1638 - writeUint64(writer, uint64(uintptr(c.Lower(value)))) 1639 1639 - } 1640 1640 - 1641 1641 - type FfiDestroyerEndpoint struct{} 1642 1642 - 1643 1643 - func (_ FfiDestroyerEndpoint) Destroy(value *Endpoint) { 1644 1644 - value.Destroy() 1645 1645 - } 1646 1646 - 1647 1294 // A filter for subscriptions and iteration. 1648 1295 type FilterInterface interface { 1649 1296 // Restrict to the global namespace, no per stream data. ··· 2559 2206 type FfiDestroyerPublicKey struct{} 2560 2207 2561 2208 func (_ FfiDestroyerPublicKey) Destroy(value *PublicKey) { 2562 2562 - value.Destroy() 2563 2563 - } 2564 2564 - 2565 2565 - type ReceiverInterface interface { 2566 2566 - NodeAddr() *NodeAddr 2567 2567 - // Subscribe to the given topic on the remote. 2568 2568 - Subscribe(remoteId *PublicKey, topic string) error 2569 2569 - // Unsubscribe from this topic on the remote. 2570 2570 - Unsubscribe(remoteId *PublicKey, topic string) error 2571 2571 - } 2572 2572 - type Receiver struct { 2573 2573 - ffiObject FfiObject 2574 2574 - } 2575 2575 - 2576 2576 - // Create a new receiver. 2577 2577 - func NewReceiver(endpoint *Endpoint, handler DataHandlerOld) (*Receiver, error) { 2578 2578 - res, err := uniffiRustCallAsync[Error]( 2579 2579 - FfiConverterErrorINSTANCE, 2580 2580 - // completeFn 2581 2581 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 2582 2582 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 2583 2583 - return res 2584 2584 - }, 2585 2585 - // liftFn 2586 2586 - func(ffi unsafe.Pointer) *Receiver { 2587 2587 - return FfiConverterReceiverINSTANCE.Lift(ffi) 2588 2588 - }, 2589 2589 - C.uniffi_iroh_streamplace_fn_constructor_receiver_new(FfiConverterEndpointINSTANCE.Lower(endpoint), FfiConverterDataHandlerOldINSTANCE.Lower(handler)), 2590 2590 - // pollFn 2591 2591 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2592 2592 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 2593 2593 - }, 2594 2594 - // freeFn 2595 2595 - func(handle C.uint64_t) { 2596 2596 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 2597 2597 - }, 2598 2598 - ) 2599 2599 - 2600 2600 - if err == nil { 2601 2601 - return res, nil 2602 2602 - } 2603 2603 - 2604 2604 - return res, err 2605 2605 - } 2606 2606 - 2607 2607 - func (_self *Receiver) NodeAddr() *NodeAddr { 2608 2608 - _pointer := _self.ffiObject.incrementPointer("*Receiver") 2609 2609 - defer _self.ffiObject.decrementPointer() 2610 2610 - res, _ := uniffiRustCallAsync[error]( 2611 2611 - nil, 2612 2612 - // completeFn 2613 2613 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 2614 2614 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 2615 2615 - return res 2616 2616 - }, 2617 2617 - // liftFn 2618 2618 - func(ffi unsafe.Pointer) *NodeAddr { 2619 2619 - return FfiConverterNodeAddrINSTANCE.Lift(ffi) 2620 2620 - }, 2621 2621 - C.uniffi_iroh_streamplace_fn_method_receiver_node_addr( 2622 2622 - _pointer), 2623 2623 - // pollFn 2624 2624 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2625 2625 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 2626 2626 - }, 2627 2627 - // freeFn 2628 2628 - func(handle C.uint64_t) { 2629 2629 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 2630 2630 - }, 2631 2631 - ) 2632 2632 - 2633 2633 - return res 2634 2634 - } 2635 2635 - 2636 2636 - // Subscribe to the given topic on the remote. 2637 2637 - func (_self *Receiver) Subscribe(remoteId *PublicKey, topic string) error { 2638 2638 - _pointer := _self.ffiObject.incrementPointer("*Receiver") 2639 2639 - defer _self.ffiObject.decrementPointer() 2640 2640 - _, err := uniffiRustCallAsync[Error]( 2641 2641 - FfiConverterErrorINSTANCE, 2642 2642 - // completeFn 2643 2643 - func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 2644 2644 - C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 2645 2645 - return struct{}{} 2646 2646 - }, 2647 2647 - // liftFn 2648 2648 - func(_ struct{}) struct{} { return struct{}{} }, 2649 2649 - C.uniffi_iroh_streamplace_fn_method_receiver_subscribe( 2650 2650 - _pointer, FfiConverterPublicKeyINSTANCE.Lower(remoteId), FfiConverterStringINSTANCE.Lower(topic)), 2651 2651 - // pollFn 2652 2652 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2653 2653 - C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 2654 2654 - }, 2655 2655 - // freeFn 2656 2656 - func(handle C.uint64_t) { 2657 2657 - C.ffi_iroh_streamplace_rust_future_free_void(handle) 2658 2658 - }, 2659 2659 - ) 2660 2660 - 2661 2661 - if err == nil { 2662 2662 - return nil 2663 2663 - } 2664 2664 - 2665 2665 - return err 2666 2666 - } 2667 2667 - 2668 2668 - // Unsubscribe from this topic on the remote. 2669 2669 - func (_self *Receiver) Unsubscribe(remoteId *PublicKey, topic string) error { 2670 2670 - _pointer := _self.ffiObject.incrementPointer("*Receiver") 2671 2671 - defer _self.ffiObject.decrementPointer() 2672 2672 - _, err := uniffiRustCallAsync[Error]( 2673 2673 - FfiConverterErrorINSTANCE, 2674 2674 - // completeFn 2675 2675 - func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 2676 2676 - C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 2677 2677 - return struct{}{} 2678 2678 - }, 2679 2679 - // liftFn 2680 2680 - func(_ struct{}) struct{} { return struct{}{} }, 2681 2681 - C.uniffi_iroh_streamplace_fn_method_receiver_unsubscribe( 2682 2682 - _pointer, FfiConverterPublicKeyINSTANCE.Lower(remoteId), FfiConverterStringINSTANCE.Lower(topic)), 2683 2683 - // pollFn 2684 2684 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2685 2685 - C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 2686 2686 - }, 2687 2687 - // freeFn 2688 2688 - func(handle C.uint64_t) { 2689 2689 - C.ffi_iroh_streamplace_rust_future_free_void(handle) 2690 2690 - }, 2691 2691 - ) 2692 2692 - 2693 2693 - if err == nil { 2694 2694 - return nil 2695 2695 - } 2696 2696 - 2697 2697 - return err 2698 2698 - } 2699 2699 - func (object *Receiver) Destroy() { 2700 2700 - runtime.SetFinalizer(object, nil) 2701 2701 - object.ffiObject.destroy() 2702 2702 - } 2703 2703 - 2704 2704 - type FfiConverterReceiver struct{} 2705 2705 - 2706 2706 - var FfiConverterReceiverINSTANCE = FfiConverterReceiver{} 2707 2707 - 2708 2708 - func (c FfiConverterReceiver) Lift(pointer unsafe.Pointer) *Receiver { 2709 2709 - result := &Receiver{ 2710 2710 - newFfiObject( 2711 2711 - pointer, 2712 2712 - func(pointer unsafe.Pointer, status *C.RustCallStatus) unsafe.Pointer { 2713 2713 - return C.uniffi_iroh_streamplace_fn_clone_receiver(pointer, status) 2714 2714 - }, 2715 2715 - func(pointer unsafe.Pointer, status *C.RustCallStatus) { 2716 2716 - C.uniffi_iroh_streamplace_fn_free_receiver(pointer, status) 2717 2717 - }, 2718 2718 - ), 2719 2719 - } 2720 2720 - runtime.SetFinalizer(result, (*Receiver).Destroy) 2721 2721 - return result 2722 2722 - } 2723 2723 - 2724 2724 - func (c FfiConverterReceiver) Read(reader io.Reader) *Receiver { 2725 2725 - return c.Lift(unsafe.Pointer(uintptr(readUint64(reader)))) 2726 2726 - } 2727 2727 - 2728 2728 - func (c FfiConverterReceiver) Lower(value *Receiver) unsafe.Pointer { 2729 2729 - // TODO: this is bad - all synchronization from ObjectRuntime.go is discarded here, 2730 2730 - // because the pointer will be decremented immediately after this function returns, 2731 2731 - // and someone will be left holding onto a non-locked pointer. 2732 2732 - pointer := value.ffiObject.incrementPointer("*Receiver") 2733 2733 - defer value.ffiObject.decrementPointer() 2734 2734 - return pointer 2735 2735 - 2736 2736 - } 2737 2737 - 2738 2738 - func (c FfiConverterReceiver) Write(writer io.Writer, value *Receiver) { 2739 2739 - writeUint64(writer, uint64(uintptr(c.Lower(value)))) 2740 2740 - } 2741 2741 - 2742 2742 - type FfiDestroyerReceiver struct{} 2743 2743 - 2744 2744 - func (_ FfiDestroyerReceiver) Destroy(value *Receiver) { 2745 2745 - value.Destroy() 2746 2746 - } 2747 2747 - 2748 2748 - type SenderInterface interface { 2749 2749 - NodeAddr() *NodeAddr 2750 2750 - // Sends the given data to all subscribers that have subscribed to this `key`. 2751 2751 - Send(key string, data []byte) error 2752 2752 - } 2753 2753 - type Sender struct { 2754 2754 - ffiObject FfiObject 2755 2755 - } 2756 2756 - 2757 2757 - // Create a new sender. 2758 2758 - func NewSender(endpoint *Endpoint) *Sender { 2759 2759 - res, _ := uniffiRustCallAsync[error]( 2760 2760 - nil, 2761 2761 - // completeFn 2762 2762 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 2763 2763 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 2764 2764 - return res 2765 2765 - }, 2766 2766 - // liftFn 2767 2767 - func(ffi unsafe.Pointer) *Sender { 2768 2768 - return FfiConverterSenderINSTANCE.Lift(ffi) 2769 2769 - }, 2770 2770 - C.uniffi_iroh_streamplace_fn_constructor_sender_new(FfiConverterEndpointINSTANCE.Lower(endpoint)), 2771 2771 - // pollFn 2772 2772 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2773 2773 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 2774 2774 - }, 2775 2775 - // freeFn 2776 2776 - func(handle C.uint64_t) { 2777 2777 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 2778 2778 - }, 2779 2779 - ) 2780 2780 - 2781 2781 - return res 2782 2782 - } 2783 2783 - 2784 2784 - func (_self *Sender) NodeAddr() *NodeAddr { 2785 2785 - _pointer := _self.ffiObject.incrementPointer("*Sender") 2786 2786 - defer _self.ffiObject.decrementPointer() 2787 2787 - res, _ := uniffiRustCallAsync[error]( 2788 2788 - nil, 2789 2789 - // completeFn 2790 2790 - func(handle C.uint64_t, status *C.RustCallStatus) unsafe.Pointer { 2791 2791 - res := C.ffi_iroh_streamplace_rust_future_complete_pointer(handle, status) 2792 2792 - return res 2793 2793 - }, 2794 2794 - // liftFn 2795 2795 - func(ffi unsafe.Pointer) *NodeAddr { 2796 2796 - return FfiConverterNodeAddrINSTANCE.Lift(ffi) 2797 2797 - }, 2798 2798 - C.uniffi_iroh_streamplace_fn_method_sender_node_addr( 2799 2799 - _pointer), 2800 2800 - // pollFn 2801 2801 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2802 2802 - C.ffi_iroh_streamplace_rust_future_poll_pointer(handle, continuation, data) 2803 2803 - }, 2804 2804 - // freeFn 2805 2805 - func(handle C.uint64_t) { 2806 2806 - C.ffi_iroh_streamplace_rust_future_free_pointer(handle) 2807 2807 - }, 2808 2808 - ) 2809 2809 - 2810 2810 - return res 2811 2811 - } 2812 2812 - 2813 2813 - // Sends the given data to all subscribers that have subscribed to this `key`. 2814 2814 - func (_self *Sender) Send(key string, data []byte) error { 2815 2815 - _pointer := _self.ffiObject.incrementPointer("*Sender") 2816 2816 - defer _self.ffiObject.decrementPointer() 2817 2817 - _, err := uniffiRustCallAsync[Error]( 2818 2818 - FfiConverterErrorINSTANCE, 2819 2819 - // completeFn 2820 2820 - func(handle C.uint64_t, status *C.RustCallStatus) struct{} { 2821 2821 - C.ffi_iroh_streamplace_rust_future_complete_void(handle, status) 2822 2822 - return struct{}{} 2823 2823 - }, 2824 2824 - // liftFn 2825 2825 - func(_ struct{}) struct{} { return struct{}{} }, 2826 2826 - C.uniffi_iroh_streamplace_fn_method_sender_send( 2827 2827 - _pointer, FfiConverterStringINSTANCE.Lower(key), FfiConverterBytesINSTANCE.Lower(data)), 2828 2828 - // pollFn 2829 2829 - func(handle C.uint64_t, continuation C.UniffiRustFutureContinuationCallback, data C.uint64_t) { 2830 2830 - C.ffi_iroh_streamplace_rust_future_poll_void(handle, continuation, data) 2831 2831 - }, 2832 2832 - // freeFn 2833 2833 - func(handle C.uint64_t) { 2834 2834 - C.ffi_iroh_streamplace_rust_future_free_void(handle) 2835 2835 - }, 2836 2836 - ) 2837 2837 - 2838 2838 - if err == nil { 2839 2839 - return nil 2840 2840 - } 2841 2841 - 2842 2842 - return err 2843 2843 - } 2844 2844 - func (object *Sender) Destroy() { 2845 2845 - runtime.SetFinalizer(object, nil) 2846 2846 - object.ffiObject.destroy() 2847 2847 - } 2848 2848 - 2849 2849 - type FfiConverterSender struct{} 2850 2850 - 2851 2851 - var FfiConverterSenderINSTANCE = FfiConverterSender{} 2852 2852 - 2853 2853 - func (c FfiConverterSender) Lift(pointer unsafe.Pointer) *Sender { 2854 2854 - result := &Sender{ 2855 2855 - newFfiObject( 2856 2856 - pointer, 2857 2857 - func(pointer unsafe.Pointer, status *C.RustCallStatus) unsafe.Pointer { 2858 2858 - return C.uniffi_iroh_streamplace_fn_clone_sender(pointer, status) 2859 2859 - }, 2860 2860 - func(pointer unsafe.Pointer, status *C.RustCallStatus) { 2861 2861 - C.uniffi_iroh_streamplace_fn_free_sender(pointer, status) 2862 2862 - }, 2863 2863 - ), 2864 2864 - } 2865 2865 - runtime.SetFinalizer(result, (*Sender).Destroy) 2866 2866 - return result 2867 2867 - } 2868 2868 - 2869 2869 - func (c FfiConverterSender) Read(reader io.Reader) *Sender { 2870 2870 - return c.Lift(unsafe.Pointer(uintptr(readUint64(reader)))) 2871 2871 - } 2872 2872 - 2873 2873 - func (c FfiConverterSender) Lower(value *Sender) unsafe.Pointer { 2874 2874 - // TODO: this is bad - all synchronization from ObjectRuntime.go is discarded here, 2875 2875 - // because the pointer will be decremented immediately after this function returns, 2876 2876 - // and someone will be left holding onto a non-locked pointer. 2877 2877 - pointer := value.ffiObject.incrementPointer("*Sender") 2878 2878 - defer value.ffiObject.decrementPointer() 2879 2879 - return pointer 2880 2880 - 2881 2881 - } 2882 2882 - 2883 2883 - func (c FfiConverterSender) Write(writer io.Writer, value *Sender) { 2884 2884 - writeUint64(writer, uint64(uintptr(c.Lower(value)))) 2885 2885 - } 2886 2886 - 2887 2887 - type FfiDestroyerSender struct{} 2888 2888 - 2889 2889 - func (_ FfiDestroyerSender) Destroy(value *Sender) { 2890 2209 value.Destroy() 2891 2210 } 2892 2211

-180

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h

··· 392 392 393 393 394 394 #endif 395 395 - #ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_DATA_HANDLER_OLD_METHOD0 396 396 - #define UNIFFI_FFIDEF_CALLBACK_INTERFACE_DATA_HANDLER_OLD_METHOD0 397 397 - typedef void (*UniffiCallbackInterfaceDataHandlerOldMethod0)(uint64_t uniffi_handle, RustBuffer topic, RustBuffer data, UniffiForeignFutureCompleteVoid uniffi_future_callback, uint64_t uniffi_callback_data, UniffiForeignFuture* uniffi_out_return); 398 398 - 399 399 - // Making function static works arround: 400 400 - // https://github.com/golang/go/issues/11263 401 401 - static void call_UniffiCallbackInterfaceDataHandlerOldMethod0( 402 402 - UniffiCallbackInterfaceDataHandlerOldMethod0 cb, uint64_t uniffi_handle, RustBuffer topic, RustBuffer data, UniffiForeignFutureCompleteVoid uniffi_future_callback, uint64_t uniffi_callback_data, UniffiForeignFuture* uniffi_out_return) 403 403 - { 404 404 - return cb(uniffi_handle, topic, data, uniffi_future_callback, uniffi_callback_data, uniffi_out_return); 405 405 - } 406 406 - 407 407 - 408 408 - #endif 409 395 #ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_GO_SIGNER_METHOD0 410 396 #define UNIFFI_FFIDEF_CALLBACK_INTERFACE_GO_SIGNER_METHOD0 411 397 typedef void (*UniffiCallbackInterfaceGoSignerMethod0)(uint64_t uniffi_handle, RustBuffer data, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); ··· 428 414 } UniffiVTableCallbackInterfaceDataHandler; 429 415 430 416 #endif 431 431 - #ifndef UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_DATA_HANDLER_OLD 432 432 - #define UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_DATA_HANDLER_OLD 433 433 - typedef struct UniffiVTableCallbackInterfaceDataHandlerOld { 434 434 - UniffiCallbackInterfaceDataHandlerOldMethod0 handleData; 435 435 - UniffiCallbackInterfaceFree uniffiFree; 436 436 - } UniffiVTableCallbackInterfaceDataHandlerOld; 437 437 - 438 438 - #endif 439 417 #ifndef UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_GO_SIGNER 440 418 #define UNIFFI_FFIDEF_V_TABLE_CALLBACK_INTERFACE_GO_SIGNER 441 419 typedef struct UniffiVTableCallbackInterfaceGoSigner { ··· 464 442 uint64_t uniffi_iroh_streamplace_fn_method_datahandler_handle_data(void* ptr, RustBuffer topic, RustBuffer data 465 443 ); 466 444 #endif 467 467 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_DATAHANDLEROLD 468 468 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_DATAHANDLEROLD 469 469 - void* uniffi_iroh_streamplace_fn_clone_datahandlerold(void* ptr, RustCallStatus *out_status 470 470 - ); 471 471 - #endif 472 472 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_DATAHANDLEROLD 473 473 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_DATAHANDLEROLD 474 474 - void uniffi_iroh_streamplace_fn_free_datahandlerold(void* ptr, RustCallStatus *out_status 475 475 - ); 476 476 - #endif 477 477 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_INIT_CALLBACK_VTABLE_DATAHANDLEROLD 478 478 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_INIT_CALLBACK_VTABLE_DATAHANDLEROLD 479 479 - void uniffi_iroh_streamplace_fn_init_callback_vtable_datahandlerold(UniffiVTableCallbackInterfaceDataHandlerOld* vtable 480 480 - ); 481 481 - #endif 482 482 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DATAHANDLEROLD_HANDLE_DATA 483 483 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DATAHANDLEROLD_HANDLE_DATA 484 484 - uint64_t uniffi_iroh_streamplace_fn_method_datahandlerold_handle_data(void* ptr, RustBuffer topic, RustBuffer data 485 485 - ); 486 486 - #endif 487 445 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_DB 488 446 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_DB 489 447 void* uniffi_iroh_streamplace_fn_clone_db(void* ptr, RustCallStatus *out_status ··· 512 470 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_WRITE 513 471 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_DB_WRITE 514 472 void* uniffi_iroh_streamplace_fn_method_db_write(void* ptr, RustBuffer secret, RustCallStatus *out_status 515 515 - ); 516 516 - #endif 517 517 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_ENDPOINT 518 518 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_ENDPOINT 519 519 - void* uniffi_iroh_streamplace_fn_clone_endpoint(void* ptr, RustCallStatus *out_status 520 520 - ); 521 521 - #endif 522 522 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_ENDPOINT 523 523 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_ENDPOINT 524 524 - void uniffi_iroh_streamplace_fn_free_endpoint(void* ptr, RustCallStatus *out_status 525 525 - ); 526 526 - #endif 527 527 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_ENDPOINT_NEW 528 528 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_ENDPOINT_NEW 529 529 - uint64_t uniffi_iroh_streamplace_fn_constructor_endpoint_new(void 530 530 - 531 531 - ); 532 532 - #endif 533 533 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_ENDPOINT_NODE_ADDR 534 534 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_ENDPOINT_NODE_ADDR 535 535 - uint64_t uniffi_iroh_streamplace_fn_method_endpoint_node_addr(void* ptr 536 473 ); 537 474 #endif 538 475 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_FILTER ··· 746 683 RustBuffer uniffi_iroh_streamplace_fn_method_publickey_uniffi_trait_display(void* ptr, RustCallStatus *out_status 747 684 ); 748 685 #endif 749 749 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_RECEIVER 750 750 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_RECEIVER 751 751 - void* uniffi_iroh_streamplace_fn_clone_receiver(void* ptr, RustCallStatus *out_status 752 752 - ); 753 753 - #endif 754 754 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_RECEIVER 755 755 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_RECEIVER 756 756 - void uniffi_iroh_streamplace_fn_free_receiver(void* ptr, RustCallStatus *out_status 757 757 - ); 758 758 - #endif 759 759 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_RECEIVER_NEW 760 760 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_RECEIVER_NEW 761 761 - uint64_t uniffi_iroh_streamplace_fn_constructor_receiver_new(void* endpoint, void* handler 762 762 - ); 763 763 - #endif 764 764 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_NODE_ADDR 765 765 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_NODE_ADDR 766 766 - uint64_t uniffi_iroh_streamplace_fn_method_receiver_node_addr(void* ptr 767 767 - ); 768 768 - #endif 769 769 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_SUBSCRIBE 770 770 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_SUBSCRIBE 771 771 - uint64_t uniffi_iroh_streamplace_fn_method_receiver_subscribe(void* ptr, void* remote_id, RustBuffer topic 772 772 - ); 773 773 - #endif 774 774 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_UNSUBSCRIBE 775 775 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_RECEIVER_UNSUBSCRIBE 776 776 - uint64_t uniffi_iroh_streamplace_fn_method_receiver_unsubscribe(void* ptr, void* remote_id, RustBuffer topic 777 777 - ); 778 778 - #endif 779 779 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_SENDER 780 780 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_SENDER 781 781 - void* uniffi_iroh_streamplace_fn_clone_sender(void* ptr, RustCallStatus *out_status 782 782 - ); 783 783 - #endif 784 784 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_SENDER 785 785 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FREE_SENDER 786 786 - void uniffi_iroh_streamplace_fn_free_sender(void* ptr, RustCallStatus *out_status 787 787 - ); 788 788 - #endif 789 789 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_SENDER_NEW 790 790 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CONSTRUCTOR_SENDER_NEW 791 791 - uint64_t uniffi_iroh_streamplace_fn_constructor_sender_new(void* endpoint 792 792 - ); 793 793 - #endif 794 794 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_SENDER_NODE_ADDR 795 795 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_SENDER_NODE_ADDR 796 796 - uint64_t uniffi_iroh_streamplace_fn_method_sender_node_addr(void* ptr 797 797 - ); 798 798 - #endif 799 799 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_SENDER_SEND 800 800 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_METHOD_SENDER_SEND 801 801 - uint64_t uniffi_iroh_streamplace_fn_method_sender_send(void* ptr, RustBuffer key, RustBuffer data 802 802 - ); 803 803 - #endif 804 686 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_SUBSCRIBERESPONSE 805 687 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_CLONE_SUBSCRIBERESPONSE 806 688 void* uniffi_iroh_streamplace_fn_clone_subscriberesponse(void* ptr, RustCallStatus *out_status ··· 1155 1037 1156 1038 ); 1157 1039 #endif 1158 1158 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DATAHANDLEROLD_HANDLE_DATA 1159 1159 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DATAHANDLEROLD_HANDLE_DATA 1160 1160 - uint16_t uniffi_iroh_streamplace_checksum_method_datahandlerold_handle_data(void 1161 1161 - 1162 1162 - ); 1163 1163 - #endif 1164 1040 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_ITER_WITH_OPTS 1165 1041 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_ITER_WITH_OPTS 1166 1042 uint16_t uniffi_iroh_streamplace_checksum_method_db_iter_with_opts(void ··· 1182 1058 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_WRITE 1183 1059 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_DB_WRITE 1184 1060 uint16_t uniffi_iroh_streamplace_checksum_method_db_write(void 1185 1185 - 1186 1186 - ); 1187 1187 - #endif 1188 1188 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_ENDPOINT_NODE_ADDR 1189 1189 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_ENDPOINT_NODE_ADDR 1190 1190 - uint16_t uniffi_iroh_streamplace_checksum_method_endpoint_node_addr(void 1191 1061 1192 1062 ); 1193 1063 #endif ··· 1329 1199 1330 1200 ); 1331 1201 #endif 1332 1332 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_NODE_ADDR 1333 1333 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_NODE_ADDR 1334 1334 - uint16_t uniffi_iroh_streamplace_checksum_method_receiver_node_addr(void 1335 1335 - 1336 1336 - ); 1337 1337 - #endif 1338 1338 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_SUBSCRIBE 1339 1339 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_SUBSCRIBE 1340 1340 - uint16_t uniffi_iroh_streamplace_checksum_method_receiver_subscribe(void 1341 1341 - 1342 1342 - ); 1343 1343 - #endif 1344 1344 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_UNSUBSCRIBE 1345 1345 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_RECEIVER_UNSUBSCRIBE 1346 1346 - uint16_t uniffi_iroh_streamplace_checksum_method_receiver_unsubscribe(void 1347 1347 - 1348 1348 - ); 1349 1349 - #endif 1350 1350 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SENDER_NODE_ADDR 1351 1351 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SENDER_NODE_ADDR 1352 1352 - uint16_t uniffi_iroh_streamplace_checksum_method_sender_node_addr(void 1353 1353 - 1354 1354 - ); 1355 1355 - #endif 1356 1356 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SENDER_SEND 1357 1357 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SENDER_SEND 1358 1358 - uint16_t uniffi_iroh_streamplace_checksum_method_sender_send(void 1359 1359 - 1360 1360 - ); 1361 1361 - #endif 1362 1202 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SUBSCRIBERESPONSE_NEXT_RAW 1363 1203 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_METHOD_SUBSCRIBERESPONSE_NEXT_RAW 1364 1204 uint16_t uniffi_iroh_streamplace_checksum_method_subscriberesponse_next_raw(void ··· 1371 1211 1372 1212 ); 1373 1213 #endif 1374 1374 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_ENDPOINT_NEW 1375 1375 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_ENDPOINT_NEW 1376 1376 - uint16_t uniffi_iroh_streamplace_checksum_constructor_endpoint_new(void 1377 1377 - 1378 1378 - ); 1379 1379 - #endif 1380 1214 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_FILTER_NEW 1381 1215 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_FILTER_NEW 1382 1216 uint16_t uniffi_iroh_streamplace_checksum_constructor_filter_new(void ··· 1419 1253 1420 1254 ); 1421 1255 #endif 1422 1422 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_RECEIVER_NEW 1423 1423 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_RECEIVER_NEW 1424 1424 - uint16_t uniffi_iroh_streamplace_checksum_constructor_receiver_new(void 1425 1425 - 1426 1426 - ); 1427 1427 - #endif 1428 1428 - #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_SENDER_NEW 1429 1429 - #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_CONSTRUCTOR_SENDER_NEW 1430 1430 - uint16_t uniffi_iroh_streamplace_checksum_constructor_sender_new(void 1431 1431 - 1432 1432 - ); 1433 1433 - #endif 1434 1256 #ifndef UNIFFI_FFIDEF_FFI_IROH_STREAMPLACE_UNIFFI_CONTRACT_VERSION 1435 1257 #define UNIFFI_FFIDEF_FFI_IROH_STREAMPLACE_UNIFFI_CONTRACT_VERSION 1436 1258 uint32_t ffi_iroh_streamplace_uniffi_contract_version(void ··· 1440 1262 1441 1263 void iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerMethod0(uint64_t uniffi_handle, RustBuffer topic, RustBuffer data, UniffiForeignFutureCompleteVoid uniffi_future_callback, uint64_t uniffi_callback_data, UniffiForeignFuture* uniffi_out_return); 1442 1264 void iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerFree(uint64_t handle); 1443 1443 - void iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldMethod0(uint64_t uniffi_handle, RustBuffer topic, RustBuffer data, UniffiForeignFutureCompleteVoid uniffi_future_callback, uint64_t uniffi_callback_data, UniffiForeignFuture* uniffi_out_return); 1444 1444 - void iroh_streamplace_cgo_dispatchCallbackInterfaceDataHandlerOldFree(uint64_t handle); 1445 1265 void iroh_streamplace_cgo_dispatchCallbackInterfaceGoSignerMethod0(uint64_t uniffi_handle, RustBuffer data, RustBuffer* uniffi_out_return, RustCallStatus* callStatus ); 1446 1266 void iroh_streamplace_cgo_dispatchCallbackInterfaceGoSignerFree(uint64_t handle); 1447 1267

-56

pkg/iroh/iroh_streamplace_test.go

··· 1 1 - package iroh_streamplace 2 2 - 3 3 - import ( 4 4 - "testing" 5 5 - 6 6 - "github.com/stretchr/testify/assert" 7 7 - 8 8 - iroh "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" 9 9 - ) 10 10 - 11 11 - type Message struct { 12 12 - topic string 13 13 - data []byte 14 14 - } 15 15 - 16 16 - type TestHandler struct { 17 17 - messages chan Message 18 18 - } 19 19 - 20 20 - func (handler TestHandler) HandleData(topic string, data []byte) { 21 21 - handler.messages <- Message{topic, data} 22 22 - } 23 23 - 24 24 - func TestBasicRoundtrip(t *testing.T) { 25 25 - ep1, err := iroh.NewEndpoint() 26 26 - assert.Nil(t, err) 27 27 - sender := iroh.NewSender(ep1) 28 28 - 29 29 - messages := make(chan Message, 5) 30 30 - handler := TestHandler{messages: messages} 31 31 - 32 32 - ep2, err := iroh.NewEndpoint() 33 33 - assert.NoError(t, err) 34 34 - receiver, err := iroh.NewReceiver(ep2, &handler) 35 35 - assert.NoError(t, err) 36 36 - 37 37 - senderAddr := sender.NodeAddr() 38 38 - senderId := senderAddr.NodeId() 39 39 - 40 40 - // subscribe 41 41 - err = receiver.Subscribe(senderId, "foo") 42 42 - assert.NoError(t, err) 43 43 - 44 44 - // send a few messages 45 45 - for i := range 5 { 46 46 - err = sender.Send("foo", []byte{byte(i), 0, 0, 0}) 47 47 - assert.NoError(t, err) 48 48 - } 49 49 - 50 50 - // make sure the receiver got them 51 51 - for i := range 5 { 52 52 - msg := <-messages 53 53 - assert.Equal(t, msg.topic, "foo") 54 54 - assert.Equal(t, msg.data, []byte{byte(i), 0, 0, 0}) 55 55 - } 56 56 - }

+3 -3

pkg/media/media.go

··· 58 58 Segment *model.Segment 59 59 Data []byte 60 60 Metadata *SegmentMetadata 61 61 + Local bool 61 62 } 62 63 63 64 func RunSelfTest(ctx context.Context) error { ··· 65 66 return SelfTest(ctx) 66 67 } 67 68 68 68 - func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer, rep replication.Replicator, mod model.Model, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer) (*MediaManager, error) { 69 69 + func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer, mod model.Model, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer) (*MediaManager, error) { 69 70 gstinit.InitGST() 70 71 err := SelfTest(ctx) 71 72 if err != nil { ··· 121 122 122 123 return &MediaManager{ 123 124 cli: cli, 124 124 - replicator: rep, 125 125 hlsRunning: map[string]*M3U8{}, 126 126 httpPipes: map[string]io.Writer{}, 127 127 model: mod, ··· 135 135 func (mm *MediaManager) HandleData(node *irohStreamplace.PublicKey, data []byte) { 136 136 r := bytes.NewReader(data) 137 137 ctx := context.Background() 138 138 - err := mm.ValidateMP4(ctx, r) 138 138 + err := mm.ValidateMP4(ctx, r, true) 139 139 if err != nil { 140 140 log.Log(ctx, "invalid incoming segment", "error", err) 141 141 }

+2 -3

pkg/media/media_test.go

··· 13 13 "stream.place/streamplace/pkg/config" 14 14 ct "stream.place/streamplace/pkg/config/configtesting" 15 15 "stream.place/streamplace/pkg/model" 16 16 - "stream.place/streamplace/pkg/replication/boring" 17 16 ) 18 17 19 18 func getFixture(name string) string { ··· 40 39 StatefulDB: nil, // Test doesn't need StatefulDB for now 41 40 Bus: bus.NewBus(), 42 41 } 43 43 - mm, err := MakeMediaManager(context.Background(), cli, nil, &boring.BoringReplicator{}, mod, bus.NewBus(), atsync) 42 42 + mm, err := MakeMediaManager(context.Background(), cli, nil, mod, bus.NewBus(), atsync) 44 43 require.NoError(t, err) 45 44 // ms, err := MakeMediaSigner(context.Background(), cli, "test-person", signer) 46 45 // require.NoError(t, err) ··· 126 125 f, err := os.Open(getFixture("sample-segment.mp4")) 127 126 require.NoError(t, err) 128 127 mm, _ := getStaticTestMediaManager(t) 129 129 - err = mm.ValidateMP4(context.Background(), f) 128 128 + err = mm.ValidateMP4(context.Background(), f, true) 130 129 require.NoError(t, err) 131 130 }

+1 -2

pkg/media/rtcrec_test.go

··· 11 11 "stream.place/streamplace/pkg/config" 12 12 "stream.place/streamplace/pkg/crypto/spkey" 13 13 "stream.place/streamplace/pkg/globalerror" 14 14 - "stream.place/streamplace/pkg/replication/boring" 15 14 "stream.place/streamplace/pkg/rtcrec" 16 15 ) 17 16 ··· 26 25 fs := cli.NewFlagSet("rtcrec-test") 27 26 err = cli.Parse(fs, []string{"--data-dir", dir, "-wide-open=true"}) 28 27 require.NoError(t, err) 29 29 - mm, err := MakeMediaManager(context.Background(), cli, nil, &boring.BoringReplicator{}, nil, nil, nil) 28 28 + mm, err := MakeMediaManager(context.Background(), cli, nil, nil, nil, nil) 30 29 require.NoError(t, err) 31 30 priv, pub, err := spkey.GenerateStreamKey() 32 31 require.NoError(t, err)

+1 -1

pkg/media/segmenter.go

··· 87 87 return 88 88 } 89 89 90 90 - err = mm.ValidateMP4(ctx, bytes.NewReader(bs)) 90 90 + err = mm.ValidateMP4(ctx, bytes.NewReader(bs), true) 91 91 if err != nil { 92 92 log.Error(ctx, "error validating segment", "error", err) 93 93 globalerror.GlobalError(err)

+3 -2

pkg/media/validate.go

··· 23 23 Cert string `json:"cert"` 24 24 } 25 25 26 26 - func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader) error { 26 26 + func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local bool) error { 27 27 ctx, span := otel.Tracer("signer").Start(ctx, "ValidateMP4") 28 28 defer span.End() 29 29 buf, err := io.ReadAll(input) ··· 82 82 return err 83 83 } 84 84 defer fd.Close() 85 85 - go mm.replicator.NewSegment(ctx, buf) 85 85 + 86 86 r := bytes.NewReader(buf) 87 87 if _, err := io.Copy(fd, r); err != nil { 88 88 return err ··· 102 102 Segment: seg, 103 103 Data: buf, 104 104 Metadata: meta, 105 105 + Local: local, 105 106 } 106 107 for _, ch := range mm.newSegmentSubs { 107 108 go func() { ch <- not }()

+2 -21

pkg/replication/iroh_replicator/iroh.go

··· 2 2 3 3 import ( 4 4 "context" 5 5 - 6 6 - "stream.place/streamplace/pkg/log" 7 7 - 8 8 - irohStreamplace "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" 9 5 ) 10 6 11 7 // IrohReplicator implements the replication mechanism using iroh 12 8 type IrohReplicator struct { 13 13 - topic string 14 14 - sender *irohStreamplace.Sender 15 9 } 16 10 17 17 - func NewIrohReplicator(ctx context.Context, ep *irohStreamplace.Endpoint, topic string) (*IrohReplicator, error) { 18 18 - sender := irohStreamplace.NewSender(ep) 11 11 + func NewIrohReplicator(ctx context.Context) (*IrohReplicator, error) { 19 12 20 20 - return &IrohReplicator{ 21 21 - topic: topic, 22 22 - sender: sender, 23 23 - }, nil 13 13 + return &IrohReplicator{}, nil 24 14 } 25 15 26 16 func (rep *IrohReplicator) NewSegment(ctx context.Context, bs []byte) { 27 27 - go func(topic string) { 28 28 - err := sendSegment(rep.sender, topic, bs) 29 29 - if err != nil { 30 30 - log.Log(ctx, "error replicating segment", "error", err) 31 31 - } 32 32 - }(rep.topic) 33 33 - } 34 17 35 35 - func sendSegment(endpoint *irohStreamplace.Sender, topic string, bs []byte) error { 36 36 - return endpoint.Send(topic, bs) 37 18 }

+86 -26

pkg/replication/iroh_replicator/kv.go

··· 1 1 package iroh_replicator 2 2 3 3 import ( 4 4 + "bytes" 4 5 "context" 5 6 "encoding/json" 6 7 "fmt" 7 8 "time" 8 9 10 10 + "github.com/bluesky-social/indigo/util" 11 11 + "golang.org/x/sync/errgroup" 9 12 "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" 10 13 "stream.place/streamplace/pkg/log" 14 14 + "stream.place/streamplace/pkg/media" 11 15 ) 12 16 13 13 - type SwarmKV struct { 14 14 - Node *iroh_streamplace.Node 15 15 - DB *iroh_streamplace.Db 16 16 - w *iroh_streamplace.WriteScope 17 17 + type IrohSwarm struct { 18 18 + Node *iroh_streamplace.Node 19 19 + DB *iroh_streamplace.Db 20 20 + w *iroh_streamplace.WriteScope 21 21 + mm *media.MediaManager 22 22 + segChan chan *media.NewSegmentNotification 23 23 + nodeId string 17 24 } 18 25 19 26 // A message saying "hey I ingested node data at this time" ··· 22 29 Time string `json:"time"` 23 30 } 24 31 25 25 - type DataHandler struct{} 26 26 - 27 27 - func (handler *DataHandler) HandleData(topic string, data []byte) { 28 28 - log.Log(context.Background(), "HandleData", "topic", topic, "data", len(data)) 29 29 - } 30 30 - 31 31 - func StartKV(ctx context.Context, tickets []string, secret []byte) (*SwarmKV, error) { 32 32 - handler := &DataHandler{} 32 32 + func NewSwarm(ctx context.Context, tickets []string, secret []byte, mm *media.MediaManager) (*IrohSwarm, error) { 33 33 ctx = log.WithLogValues(ctx, "func", "StartKV") 34 34 35 35 log.Log(ctx, "Starting with tickets", "tickets", tickets) ··· 39 39 MaxSendDuration: 1000_000_000, // 1s 40 40 } 41 41 log.Log(ctx, "Config created", "config", config) 42 42 - node, err := iroh_streamplace.NodeReceiver(config, handler) 42 42 + 43 43 + swarm := IrohSwarm{ 44 44 + mm: mm, 45 45 + } 46 46 + 47 47 + node, err := iroh_streamplace.NodeReceiver(config, &swarm) 43 48 if err != nil { 44 49 return nil, fmt.Errorf("failed to create NodeSender: %w", err) 45 50 } ··· 47 52 db := node.Db() 48 53 w := node.NodeScope() 49 54 50 50 - node_id, err := node.NodeId() 55 55 + swarm.DB = db 56 56 + swarm.w = w 57 57 + swarm.Node = node 58 58 + 59 59 + nodeId, err := node.NodeId() 51 60 if err != nil { 52 61 return nil, fmt.Errorf("failed to get NodeId: %w", err) 53 62 } 54 54 - log.Log(ctx, "Node ID:", "node_id", node_id) 63 63 + log.Log(ctx, "Node ID:", "node_id", nodeId) 64 64 + swarm.nodeId = nodeId.String() 55 65 56 66 ticket, err := node.Ticket() 57 67 if err != nil { ··· 59 69 } 60 70 log.Log(ctx, "Ticket:", "ticket", ticket) 61 71 62 62 - swarm := SwarmKV{ 63 63 - Node: node, 64 64 - DB: db, 65 65 - w: w, 66 66 - } 67 72 return &swarm, nil 68 73 } 69 74 70 75 var activeSubs = make(map[string]bool) 71 76 72 72 - func (swarm *SwarmKV) Start(ctx context.Context, tickets []string) error { 77 77 + func (swarm *IrohSwarm) Start(ctx context.Context, tickets []string) error { 73 78 if len(tickets) > 0 { 74 79 err := swarm.Node.JoinPeers(tickets) 75 80 if err != nil { ··· 84 89 nodeIdStr := nodeId.String() 85 90 log.Log(ctx, "Node ID:", "node_id", nodeIdStr) 86 91 92 92 + g, ctx := errgroup.WithContext(ctx) 93 93 + g.Go(func() error { 94 94 + return swarm.startKV(ctx) 95 95 + }) 96 96 + g.Go(func() error { 97 97 + return swarm.startSegmentSender(ctx) 98 98 + }) 99 99 + return g.Wait() 100 100 + } 101 101 + 102 102 + func (swarm *IrohSwarm) startKV(ctx context.Context) error { 87 103 sub := swarm.DB.Subscribe(iroh_streamplace.NewFilter()) 88 104 for { 89 105 if ctx.Err() != nil { ··· 110 126 continue 111 127 } 112 128 if !activeSubs[keyStr] { 113 113 - if info.NodeID == nodeIdStr { 129 129 + if info.NodeID == swarm.nodeId { 114 130 activeSubs[keyStr] = true 115 131 continue 116 132 } ··· 137 153 } 138 154 } 139 155 140 140 - func (swarm *SwarmKV) Put(ctx context.Context, key string, value []byte) error { 141 141 - // streamerBs := []byte(streamer) 142 142 - keyBs := []byte(key) 143 143 - return swarm.w.Put(nil, keyBs, value) 156 156 + func (swarm *IrohSwarm) startSegmentSender(ctx context.Context) error { 157 157 + ch := swarm.mm.NewSegment() 158 158 + for { 159 159 + select { 160 160 + case <-ctx.Done(): 161 161 + return ctx.Err() 162 162 + case not := <-ch: 163 163 + err := swarm.SendSegment(ctx, not) 164 164 + if err != nil { 165 165 + log.Error(ctx, "could not send segment to swarm", "error", err) 166 166 + } 167 167 + continue 168 168 + } 169 169 + } 170 170 + } 171 171 + 172 172 + func (swarm *IrohSwarm) HandleData(topic string, data []byte) { 173 173 + err := swarm.mm.ValidateMP4(context.Background(), bytes.NewReader(data), false) 174 174 + if err != nil { 175 175 + log.Error(context.Background(), "could not validate segment", "error", err) 176 176 + } 177 177 + } 178 178 + 179 179 + func (swarm *IrohSwarm) SendSegment(ctx context.Context, not *media.NewSegmentNotification) error { 180 180 + if !not.Local { 181 181 + return nil 182 182 + } 183 183 + originInfo := OriginInfo{ 184 184 + NodeID: swarm.nodeId, 185 185 + Time: not.Segment.StartTime.Format(util.ISO8601), 186 186 + } 187 187 + bs, err := json.Marshal(originInfo) 188 188 + if err != nil { 189 189 + log.Error(ctx, "could not marshal origin info", "error", err) 190 190 + return err 191 191 + } 192 192 + keyBs := []byte(not.Segment.RepoDID) 193 193 + err = swarm.w.Put(nil, keyBs, bs) 194 194 + if err != nil { 195 195 + log.Error(ctx, "could not put segment to swarm", "error", err) 196 196 + return err 197 197 + } 198 198 + err = swarm.Node.SendSegment(not.Segment.RepoDID, not.Data) 199 199 + if err != nil { 200 200 + log.Error(ctx, "could not send segment to swarm", "error", err) 201 201 + return err 202 202 + } 203 203 + return nil 144 204 }

-226

rust/iroh-streamplace/src/api.rs

··· 1 1 - //! Protocol API 2 2 - 3 3 - use std::collections::{BTreeMap, BTreeSet}; 4 4 - 5 5 - use bytes::Bytes; 6 6 - use iroh::{Endpoint, NodeId, protocol::ProtocolHandler}; 7 7 - use irpc::{Client, WithChannels, channel::oneshot, rpc::RemoteService, rpc_requests}; 8 8 - use irpc_iroh::{IrohProtocol, IrohRemoteConnection}; 9 9 - use n0_future::future::Boxed; 10 10 - use serde::{Deserialize, Serialize}; 11 11 - use tracing::{debug, warn}; 12 12 - 13 13 - /// Subscribe to the given `key` 14 14 - #[derive(Debug, Serialize, Deserialize)] 15 15 - struct Subscribe { 16 16 - key: String, 17 17 - // TODO: verify 18 18 - remote_id: NodeId, 19 19 - } 20 20 - 21 21 - /// Unsubscribe from the given `key` 22 22 - #[derive(Debug, Serialize, Deserialize)] 23 23 - struct Unsubscribe { 24 24 - key: String, 25 25 - // TODO: verify 26 26 - remote_id: NodeId, 27 27 - } 28 28 - 29 29 - #[derive(Debug, Serialize, Deserialize)] 30 30 - struct SendSegment { 31 31 - key: String, 32 32 - data: Bytes, 33 33 - } 34 34 - 35 35 - #[derive(Debug, Clone, Serialize, Deserialize)] 36 36 - struct RecvSegment { 37 37 - key: String, 38 38 - data: Bytes, 39 39 - } 40 40 - 41 41 - // Use the macro to generate both the Protocol and Message enums 42 42 - // plus implement Channels for each type 43 43 - #[rpc_requests(message = Message)] 44 44 - #[derive(Serialize, Deserialize, Debug)] 45 45 - enum Protocol { 46 46 - #[rpc(tx=oneshot::Sender<()>)] 47 47 - Subscribe(Subscribe), 48 48 - #[rpc(tx=oneshot::Sender<()>)] 49 49 - Unsubscribe(Unsubscribe), 50 50 - #[rpc(tx=oneshot::Sender<()>)] 51 51 - SendSegment(SendSegment), 52 52 - #[rpc(tx=oneshot::Sender<()>)] 53 53 - RecvSegment(RecvSegment), 54 54 - } 55 55 - 56 56 - struct Actor { 57 57 - endpoint: iroh::Endpoint, 58 58 - recv: tokio::sync::mpsc::Receiver<Message>, 59 59 - subscriptions: BTreeMap<String, BTreeSet<NodeId>>, 60 60 - connections: BTreeMap<NodeId, Connection>, 61 61 - handler: Box<dyn Fn(String, Vec<u8>) -> Boxed<()> + Send + Sync + 'static>, 62 62 - } 63 63 - 64 64 - #[derive(Debug)] 65 65 - struct Connection { 66 66 - _id: NodeId, 67 67 - rpc: Client<Protocol>, 68 68 - } 69 69 - 70 70 - impl Actor { 71 71 - fn spawn( 72 72 - endpoint: &iroh::Endpoint, 73 73 - handler: impl Fn(String, Vec<u8>) -> Boxed<()> + Send + Sync + 'static, 74 74 - ) -> Api { 75 75 - let (tx, rx) = tokio::sync::mpsc::channel(1); 76 76 - let actor = Self { 77 77 - endpoint: endpoint.clone(), 78 78 - recv: rx, 79 79 - subscriptions: BTreeMap::new(), 80 80 - connections: BTreeMap::new(), 81 81 - handler: Box::new(handler), 82 82 - }; 83 83 - n0_future::task::spawn(actor.run()); 84 84 - Api { 85 85 - inner: Client::local(tx), 86 86 - } 87 87 - } 88 88 - 89 89 - async fn run(mut self) { 90 90 - while let Some(msg) = self.recv.recv().await { 91 91 - self.handle(msg).await; 92 92 - } 93 93 - } 94 94 - 95 95 - async fn handle(&mut self, msg: Message) { 96 96 - match msg { 97 97 - Message::Subscribe(sub) => { 98 98 - debug!("subscribe {:?}", sub); 99 99 - let WithChannels { tx, inner, .. } = sub; 100 100 - 101 101 - self.subscriptions 102 102 - .entry(inner.key) 103 103 - .or_default() 104 104 - .insert(inner.remote_id); 105 105 - 106 106 - tx.send(()).await.ok(); 107 107 - } 108 108 - Message::Unsubscribe(sub) => { 109 109 - debug!("unsubscribe {:?}", sub); 110 110 - let WithChannels { tx, inner, .. } = sub; 111 111 - 112 112 - if let Some(e) = self.subscriptions.get_mut(&inner.key) { 113 113 - e.remove(&inner.remote_id); 114 114 - } 115 115 - 116 116 - tx.send(()).await.ok(); 117 117 - } 118 118 - Message::SendSegment(segment) => { 119 119 - debug!("send segment {:?}", segment); 120 120 - let WithChannels { tx, inner, .. } = segment; 121 121 - 122 122 - let msg = RecvSegment { 123 123 - key: inner.key.clone(), 124 124 - data: inner.data.clone(), 125 125 - }; 126 126 - 127 127 - for (key, remotes) in &self.subscriptions { 128 128 - if key == &inner.key { 129 129 - for remote in remotes { 130 130 - debug!("sending to topic {}: {}", key, remote); 131 131 - 132 132 - // ensure connection 133 133 - if !self.connections.contains_key(remote) { 134 134 - let conn = IrohRemoteConnection::new( 135 135 - self.endpoint.clone(), 136 136 - (*remote).into(), 137 137 - Api::ALPN.to_vec(), 138 138 - ); 139 139 - 140 140 - let conn = Connection { 141 141 - rpc: Client::boxed(conn), 142 142 - _id: *remote, 143 143 - }; 144 144 - self.connections.insert(*remote, conn); 145 145 - } 146 146 - let conn = self.connections.get(remote).expect("just checked"); 147 147 - 148 148 - if let Err(err) = conn.rpc.rpc(msg.clone()).await { 149 149 - warn!("failed to send to {}: {:?}", remote, err); 150 150 - // remove conn 151 151 - self.connections.remove(remote); 152 152 - } 153 153 - } 154 154 - } 155 155 - } 156 156 - 157 157 - tx.send(()).await.ok(); 158 158 - } 159 159 - Message::RecvSegment(segment) => { 160 160 - debug!("recv segment {:?}", segment); 161 161 - let WithChannels { tx, inner, .. } = segment; 162 162 - (self.handler)(inner.key, inner.data.to_vec()).await; 163 163 - tx.send(()).await.ok(); 164 164 - } 165 165 - } 166 166 - } 167 167 - } 168 168 - 169 169 - /// The actual API to interact with 170 170 - pub(crate) struct Api { 171 171 - inner: Client<Protocol>, 172 172 - } 173 173 - 174 174 - impl Api { 175 175 - pub(crate) const ALPN: &[u8] = b"/iroh/streamplace/1"; 176 176 - 177 177 - pub(crate) fn spawn(endpoint: &iroh::Endpoint) -> Self { 178 178 - Actor::spawn(endpoint, |_, _| Box::pin(async move {})) 179 179 - } 180 180 - 181 181 - pub(crate) fn spawn_with_handler( 182 182 - endpoint: &iroh::Endpoint, 183 183 - handler: impl Fn(String, Vec<u8>) -> Boxed<()> + Send + Sync + 'static, 184 184 - ) -> Self { 185 185 - Actor::spawn(endpoint, handler) 186 186 - } 187 187 - 188 188 - pub(crate) fn connect(endpoint: Endpoint, addr: impl Into<iroh::NodeAddr>) -> Api { 189 189 - let conn = IrohRemoteConnection::new(endpoint, addr.into(), Self::ALPN.to_vec()); 190 190 - Api { 191 191 - inner: Client::boxed(conn), 192 192 - } 193 193 - } 194 194 - 195 195 - pub(crate) fn expose(&self) -> impl ProtocolHandler { 196 196 - let local = self 197 197 - .inner 198 198 - .as_local() 199 199 - .expect("can not listen on remote service"); 200 200 - IrohProtocol::new(Protocol::remote_handler(local)) 201 201 - } 202 202 - 203 203 - pub(crate) async fn subscribe(&self, key: String, self_id: NodeId) -> irpc::Result<()> { 204 204 - self.inner 205 205 - .rpc(Subscribe { 206 206 - key, 207 207 - remote_id: self_id, 208 208 - }) 209 209 - .await 210 210 - } 211 211 - 212 212 - pub(crate) async fn unsubscribe(&self, key: String, self_id: NodeId) -> irpc::Result<()> { 213 213 - self.inner 214 214 - .rpc(Unsubscribe { 215 215 - key, 216 216 - remote_id: self_id, 217 217 - }) 218 218 - .await 219 219 - } 220 220 - 221 221 - /// Send this segment to all subscriptions. 222 222 - pub(crate) async fn send_segment(&self, key: String, data: Bytes) -> irpc::Result<()> { 223 223 - let msg = SendSegment { key, data }; 224 224 - self.inner.rpc(msg).await 225 225 - } 226 226 - }

-30

rust/iroh-streamplace/src/endpoint.rs

··· 1 1 - use iroh::Watcher; 2 2 - 3 3 - use crate::{error::Error, node_addr::NodeAddr}; 4 4 - 5 5 - #[derive(uniffi::Object, Debug, Clone)] 6 6 - pub struct Endpoint { 7 7 - pub(crate) endpoint: iroh::Endpoint, 8 8 - } 9 9 - 10 10 - #[uniffi::export] 11 11 - impl Endpoint { 12 12 - /// Create a new endpoint. 13 13 - #[uniffi::constructor(async_runtime = "tokio")] 14 14 - pub async fn new() -> Result<Self, Error> { 15 15 - let endpoint = iroh::Endpoint::builder() 16 16 - .discovery_n0() 17 17 - .discovery_local_network() 18 18 - .bind() 19 19 - .await?; 20 20 - 21 21 - Ok(Self { endpoint }) 22 22 - } 23 23 - 24 24 - #[uniffi::method(async_runtime = "tokio")] 25 25 - pub async fn node_addr(&self) -> NodeAddr { 26 26 - let _ = self.endpoint.home_relay().initialized().await; 27 27 - let addr = self.endpoint.node_addr().initialized().await; 28 28 - addr.into() 29 29 - } 30 30 - }

+1 -6

rust/iroh-streamplace/src/lib.rs

··· 1 1 uniffi::setup_scaffolding!(); 2 2 3 3 pub mod c2pa; 4 4 - pub mod endpoint; 5 4 pub mod error; 6 6 - pub mod public_key; 7 5 pub mod node; 8 8 - pub mod receiver; 9 9 - pub mod sender; 10 6 pub mod node_addr; 11 11 - 12 12 - mod api; 7 7 + pub mod public_key;

-150

rust/iroh-streamplace/src/receiver.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use iroh::protocol::Router; 4 4 - 5 5 - use crate::{api::Api, endpoint::Endpoint, error::Error, public_key::PublicKey, node_addr::NodeAddr}; 6 6 - 7 7 - #[derive(uniffi::Object)] 8 8 - pub struct Receiver { 9 9 - endpoint: Endpoint, 10 10 - _api: Api, 11 11 - _router: iroh::protocol::Router, 12 12 - } 13 13 - 14 14 - #[uniffi::export] 15 15 - impl Receiver { 16 16 - /// Create a new receiver. 17 17 - #[uniffi::constructor(async_runtime = "tokio")] 18 18 - pub async fn new( 19 19 - endpoint: &Endpoint, 20 20 - handler: Arc<dyn DataHandlerOld>, 21 21 - ) -> Result<Receiver, Error> { 22 22 - let api = Api::spawn_with_handler(&endpoint.endpoint, move |id, data| { 23 23 - let handler = handler.clone(); 24 24 - Box::pin(async move { 25 25 - handler.handle_data(id, data).await; 26 26 - }) 27 27 - }); 28 28 - let router = Router::builder(endpoint.endpoint.clone()) 29 29 - .accept(Api::ALPN, api.expose()) 30 30 - .spawn(); 31 31 - 32 32 - Ok(Receiver { 33 33 - endpoint: endpoint.clone(), 34 34 - _api: api, 35 35 - _router: router, 36 36 - }) 37 37 - } 38 38 - 39 39 - /// Subscribe to the given topic on the remote. 40 40 - #[uniffi::method(async_runtime = "tokio")] 41 41 - pub async fn subscribe(&self, remote_id: Arc<PublicKey>, topic: &str) -> Result<(), Error> { 42 42 - let remote_id: iroh::NodeId = remote_id.as_ref().into(); 43 43 - let api = Api::connect(self.endpoint.endpoint.clone(), remote_id); 44 44 - api.subscribe(topic.to_string(), self.endpoint.endpoint.node_id()) 45 45 - .await?; 46 46 - Ok(()) 47 47 - } 48 48 - 49 49 - /// Unsubscribe from this topic on the remote. 50 50 - #[uniffi::method(async_runtime = "tokio")] 51 51 - pub async fn unsubscribe( 52 52 - &self, 53 53 - remote_id: Arc<PublicKey>, 54 54 - topic: &str, 55 55 - ) -> Result<(), Error> { 56 56 - let remote_id: iroh::NodeId = remote_id.as_ref().into(); 57 57 - let api = Api::connect(self.endpoint.endpoint.clone(), remote_id); 58 58 - api.unsubscribe(topic.to_string(), self.endpoint.endpoint.node_id()) 59 59 - .await?; 60 60 - Ok(()) 61 61 - } 62 62 - 63 63 - #[uniffi::method(async_runtime = "tokio")] 64 64 - pub async fn node_addr(&self) -> NodeAddr { 65 65 - self.endpoint.node_addr().await 66 66 - } 67 67 - } 68 68 - 69 69 - #[uniffi::export(with_foreign)] 70 70 - #[async_trait::async_trait] 71 71 - pub trait DataHandlerOld: Send + Sync { 72 72 - async fn handle_data(&self, topic: String, data: Vec<u8>); 73 73 - } 74 74 - 75 75 - #[cfg(test)] 76 76 - mod tests { 77 77 - 78 78 - use super::*; 79 79 - use crate::sender::Sender; 80 80 - 81 81 - #[tokio::test] 82 82 - async fn test_roundtrip() { 83 83 - tracing_subscriber::fmt() 84 84 - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) 85 85 - .init(); 86 86 - 87 87 - let ep1 = Endpoint::new().await.unwrap(); 88 88 - let sender = Sender::new(&ep1).await.unwrap(); 89 89 - 90 90 - let (s, mut r) = tokio::sync::mpsc::channel(5); 91 91 - 92 92 - #[derive(Debug, Clone)] 93 93 - struct TestHandler { 94 94 - messages: tokio::sync::mpsc::Sender<(String, Vec<u8>)>, 95 95 - } 96 96 - 97 97 - #[async_trait::async_trait] 98 98 - impl DataHandlerOld for TestHandler { 99 99 - async fn handle_data(&self, topic: String, data: Vec<u8>) { 100 100 - self.messages.send((topic, data)).await.unwrap(); 101 101 - } 102 102 - } 103 103 - 104 104 - let handler = TestHandler { messages: s }; 105 105 - let ep2 = Endpoint::new().await.unwrap(); 106 106 - let receiver = Receiver::new(&ep2, Arc::new(handler.clone())) 107 107 - .await 108 108 - .unwrap(); 109 109 - 110 110 - let sender_addr = sender.node_addr().await; 111 111 - println!("sender addr: {sender_addr:?}"); 112 112 - 113 113 - let receiver_addr = receiver.node_addr().await; 114 114 - println!("recv addr: {receiver_addr:?}"); 115 115 - 116 116 - // subscribe 117 117 - receiver 118 118 - .subscribe(Arc::new(sender_addr.node_id()), "foo") 119 119 - .await 120 120 - .unwrap(); 121 121 - 122 122 - // send a few messages 123 123 - for i in 0u8..5 { 124 124 - sender.send("foo", &[i, 0, 0, 0]).await.unwrap(); 125 125 - } 126 126 - 127 127 - // make sure the receiver got them 128 128 - for i in 0u8..5 { 129 129 - let (topic, msg) = r.recv().await.unwrap(); 130 130 - assert_eq!(topic, "foo"); 131 131 - assert_eq!(msg, vec![i, 0, 0, 0]); 132 132 - } 133 133 - 134 134 - // unsubscribe 135 135 - receiver 136 136 - .unsubscribe(Arc::new(sender_addr.node_id()), "foo") 137 137 - .await 138 138 - .unwrap(); 139 139 - 140 140 - // send a message, shouldn't error 141 141 - sender.send("foo", &[1]).await.unwrap(); 142 142 - 143 143 - // no message received, times out 144 144 - let res = tokio::time::timeout(std::time::Duration::from_millis(200), async { 145 145 - r.recv().await.unwrap(); 146 146 - }) 147 147 - .await; 148 148 - assert!(res.is_err()); 149 149 - } 150 150 - }

-43

rust/iroh-streamplace/src/sender.rs

··· 1 1 - use bytes::Bytes; 2 2 - use iroh::protocol::Router; 3 3 - 4 4 - use crate::{api::Api, c2pa::SPError, endpoint::Endpoint, error::Error, node_addr::NodeAddr}; 5 5 - 6 6 - #[derive(uniffi::Object)] 7 7 - pub struct Sender { 8 8 - endpoint: Endpoint, 9 9 - api: Api, 10 10 - _router: iroh::protocol::Router, 11 11 - } 12 12 - 13 13 - #[uniffi::export] 14 14 - impl Sender { 15 15 - /// Create a new sender. 16 16 - #[uniffi::constructor(async_runtime = "tokio")] 17 17 - pub async fn new(endpoint: &Endpoint) -> Sender { 18 18 - let api = Api::spawn(&endpoint.endpoint); 19 19 - let router = Router::builder(endpoint.endpoint.clone()) 20 20 - .accept(Api::ALPN, api.expose()) 21 21 - .spawn(); 22 22 - 23 23 - Sender { 24 24 - endpoint: endpoint.clone(), 25 25 - api, 26 26 - _router: router, 27 27 - } 28 28 - } 29 29 - 30 30 - /// Sends the given data to all subscribers that have subscribed to this `key`. 31 31 - #[uniffi::method(async_runtime = "tokio")] 32 32 - pub async fn send(&self, key: &str, data: &[u8]) -> Result<(), Error> { 33 33 - self.api 34 34 - .send_segment(key.to_string(), Bytes::copy_from_slice(data)) 35 35 - .await?; 36 36 - Ok(()) 37 37 - } 38 38 - 39 39 - #[uniffi::method(async_runtime = "tokio")] 40 40 - pub async fn node_addr(&self) -> NodeAddr { 41 41 - self.endpoint.node_addr().await 42 42 - } 43 43 - }