@@ -2403,3 +2403,78 @@ func TestCreateOrUpdateStreamCrossDomains(t *testing.T) {
2403
2403
t .Fatalf ("Expected subject to be updated to 'bar', got %q" , info2 .Config .Subjects [0 ])
2404
2404
}
2405
2405
}
2406
+
2407
+ func TestPromoteMirrorToStream (t * testing.T ) {
2408
+ srv := RunBasicJetStreamServer ()
2409
+ defer shutdownJSServerAndRemoveStorage (t , srv )
2410
+ nc , err := nats .Connect (srv .ClientURL ())
2411
+ if err != nil {
2412
+ t .Fatalf ("Unexpected error: %v" , err )
2413
+ }
2414
+ defer nc .Close ()
2415
+
2416
+ js , err := jetstream .New (nc )
2417
+ if err != nil {
2418
+ t .Fatalf ("Unexpected error: %v" , err )
2419
+ }
2420
+
2421
+ ctx := context .Background ()
2422
+ // Create source stream
2423
+ _ , err = js .CreateStream (ctx , jetstream.StreamConfig {
2424
+ Name : "SOURCE" ,
2425
+ Subjects : []string {"foo" },
2426
+ })
2427
+ if err != nil {
2428
+ t .Fatalf ("Unexpected error: %v" , err )
2429
+ }
2430
+
2431
+ // Publish some messages to source
2432
+ for range 10 {
2433
+ if _ , err := js .Publish (ctx , "foo" , []byte ("hello" )); err != nil {
2434
+ t .Fatalf ("Unexpected error: %v" , err )
2435
+ }
2436
+ }
2437
+
2438
+ // Create mirror stream
2439
+ mirror , err := js .CreateStream (ctx , jetstream.StreamConfig {
2440
+ Name : "MIRROR" ,
2441
+ Mirror : & jetstream.StreamSource {Name : "SOURCE" },
2442
+ })
2443
+ if err != nil {
2444
+ t .Fatalf ("Unexpected error: %v" , err )
2445
+ }
2446
+
2447
+ // Wait for mirror to catch up
2448
+ checkFor (t , 5 * time .Second , 100 * time .Millisecond , func () error {
2449
+ info , err := mirror .Info (ctx )
2450
+ if err != nil {
2451
+ return err
2452
+ }
2453
+ if info .State .Msgs != 10 {
2454
+ return fmt .Errorf ("not caught up" )
2455
+ }
2456
+ return nil
2457
+ })
2458
+
2459
+ // Remove mirror configuration
2460
+ updatedMirror , err := js .UpdateStream (ctx , jetstream.StreamConfig {
2461
+ Name : "MIRROR" ,
2462
+ Subjects : []string {"bar" },
2463
+ })
2464
+ if err != nil {
2465
+ t .Fatalf ("Unexpected error: %v" , err )
2466
+ }
2467
+
2468
+ info , err := updatedMirror .Info (ctx )
2469
+ if err != nil {
2470
+ t .Fatalf ("Unexpected error: %v" , err )
2471
+ }
2472
+
2473
+ if info .Config .Mirror != nil {
2474
+ t .Fatal ("Expected mirror to be nil after promote" )
2475
+ }
2476
+
2477
+ if info .Config .Subjects [0 ] != "bar" {
2478
+ t .Fatalf ("Expected subject to be 'bar', got %q" , info .Config .Subjects [0 ])
2479
+ }
2480
+ }
0 commit comments