import "github.com/xinjiayu/rxgo"

package main

import (
    "fmt"
    "github.com/xinjiayu/rxgo"
)

func main() {
    // 创建Observable
    observable := rxgo.Just(1, 2, 3, 4, 5)
    
    // 应用操作符链
    result := observable.
        Map(func(value interface{}) (interface{}, error) {
            return value.(int) * 2, nil
        }).
        Filter(func(value interface{}) bool {
            return value.(int) > 5
        }).
        Take(3)
    
    // 订阅并处理结果
    result.SubscribeWithCallbacks(
        func(value interface{}) {
            fmt.Printf("接收到: %v\n", value)
        },
        func(err error) {
            fmt.Printf("错误: %v\n", err)
        },
        func() {
            fmt.Println("完成")
        },
    )
}
// 输出:
// 接收到: 6
// 接收到: 8
// 接收到: 10
// 完成

// PublishSubject - 实时广播
subject := rxgo.NewPublishSubject()

// 多个订阅者
sub1 := subject.SubscribeWithCallbacks(onNext1, onError1, onComplete1)
sub2 := subject.SubscribeWithCallbacks(onNext2, onError2, onComplete2)

// 发送数据
subject.OnNext(1)
subject.OnNext(2)
subject.OnComplete()

// BehaviorSubject - 状态流
behaviorSubject := rxgo.NewBehaviorSubject(0) // 初始值为0
behaviorSubject.OnNext(1)
// 新订阅者会立即收到最新值(1)
lateSub := behaviorSubject.Subscribe(observer)

// 创建支持背压的Flowable
flowable := rxgo.FlowableCreate(func(subscriber rxgo.Subscriber) {
    // 生产大量数据
    for i := 0; i < 1000000; i++ {
        subscriber.OnNext(rxgo.CreateItem(i))
    }
    subscriber.OnComplete()
})

// 背压策略
buffered := flowable.OnBackpressureBuffer()           // 缓冲策略
dropped := flowable.OnBackpressureDrop()              // 丢弃策略  
latest := flowable.OnBackpressureLatest()             // 保留最新策略

// 订阅时控制请求速率
subscription := buffered.SubscribeWithCallbacks(onNext, onError, onComplete)
subscription.Request(10) // 请求10个元素

// 创建并行Flowable
parallel := rxgo.ParallelRange(1, 1000, 4). // 4个并行分区
    Map(func(v interface{}) (interface{}, error) {
        // CPU密集型操作
        return complexCalculation(v.(int)), nil
    }).
    Filter(func(v interface{}) bool {
        return v.(int) > 100
    })

// 合并结果
single := parallel.Reduce(func(acc, curr interface{}) interface{} {
    return acc.(int) + curr.(int)
})

result, err := single.BlockingGet()

// Join操作符 - 基于时间窗口的连接
left := rxgo.Interval(100 * time.Millisecond).Take(5)
right := rxgo.Interval(150 * time.Millisecond).Take(5)

joined := left.Join(
    right,
    func(item interface{}) time.Duration { return 200 * time.Millisecond }, // 左窗口
    func(item interface{}) time.Duration { return 200 * time.Millisecond }, // 右窗口
    func(l, r interface{}) interface{} { return fmt.Sprintf("L%v-R%v", l, r) },
)

// WithLatestFrom - 组合最新值
source := rxgo.Interval(100 * time.Millisecond)
other := rxgo.Interval(200 * time.Millisecond)

combined := source.WithLatestFrom(other, 
    func(s, o interface{}) interface{} {
        return fmt.Sprintf("source:%v, latest:%v", s, o)
    })

// 立即调度器 - 在当前goroutine中执行
observable.SubscribeOn(rxgo.ImmediateScheduler)

// 新线程调度器 - 每次创建新goroutine
observable.SubscribeOn(rxgo.NewThreadScheduler)

// 线程池调度器 - 使用goroutine池
observable.SubscribeOn(rxgo.ThreadPoolScheduler)

// 当前线程调度器 - 队列执行
observable.SubscribeOn(rxgo.CurrentThreadScheduler)

// 测试调度器 - 虚拟时间
testScheduler := rxgo.NewTestScheduler()
observable.SubscribeOn(testScheduler)

// 创建工作窃取调度器
wsScheduler := rxgo.NewWorkStealingScheduler(runtime.NumCPU())
wsScheduler.Start()

// 使用工作窃取调度器
observable.SubscribeOn(wsScheduler)

// 获取性能统计
stats := rxgo.GetWorkStealingStats()
fmt.Printf("负载均衡比例: %.2f\n", stats.LoadBalanceRatio)
fmt.Printf("工作窃取次数: %d\n", stats.WorkStealingCount)

// 使用对象池
item := rxgo.GetPooledItem()
item.Value = data
defer rxgo.ReturnPooledItem(item)

// 获取性能统计
stats := rxgo.GetPerformanceStats()
fmt.Printf("内存效率: %.1f%%\n", stats.MemoryEfficiency)
fmt.Printf("GC压力减少: %d%%\n", stats.GCReduction)

// 标量优化 - Just操作符在编译时优化
scalar := rxgo.Just(42).Map(func(v interface{}) (interface{}, error) {
    return v.(int) * 2, nil // 编译时直接计算为84
})

// 空Observable优化
empty := rxgo.Empty().Map(transform) // 直接返回Empty，无运行时开销

// ConditionalSubscriber - 微融合优化
observable := rxgo.Range(1, 1000000).
    Filter(func(v interface{}) bool { return v.(int)%2 == 0 }).
    Map(func(v interface{}) (interface{}, error) { return v.(int)*2, nil }).
    Take(1000)
// 运行时会自动融合过滤和映射操作，减少中间分配

// 超时控制
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

observable := rxgo.Interval(100*time.Millisecond).WithContext(ctx)
// 5秒后自动取消

// 手动取消
ctx, cancel := context.WithCancel(context.Background())
observable := rxgo.Range(1, 1000000).WithContext(ctx)
go func() {
    time.Sleep(1*time.Second)
    cancel() // 1秒后取消处理
}()

observable := rxgo.Create(func(observer rxgo.Observer) {
    observer(rxgo.CreateItem(1))
    observer(rxgo.CreateErrorItem(errors.New("something wrong")))
}).
Catch(func(err error) rxgo.Observable {
    return rxgo.Just("fallback") // 错误恢复
}).
Retry(3). // 重试3次
RetryWhen(func(errors rxgo.Observable) rxgo.Observable {
    return errors.Delay(1*time.Second) // 延迟重试
})

// 创建自定义操作符
func CustomBatch(size int) func(rxgo.Observable) rxgo.Observable {
    return func(source rxgo.Observable) rxgo.Observable {
        return rxgo.Create(func(observer rxgo.Observer) {
            batch := make([]interface{}, 0, size)
            
            source.Subscribe(func(item rxgo.Item) {
                if item.IsError() {
                    observer(item)
                    return
                }
                
                if item.Value == nil {
                    if len(batch) > 0 {
                        observer(rxgo.CreateItem(batch))
                    }
                    observer(item)
                    return
                }
                
                batch = append(batch, item.Value)
                if len(batch) >= size {
                    observer(rxgo.CreateItem(batch))
                    batch = batch[:0]
                }
            })
        })
    }
}

// 使用自定义操作符
observable.Transform(CustomBatch(10))

func TestWithVirtualTime(t *testing.T) {
    scheduler := rxgo.NewTestScheduler()
    
    // 创建虚拟时间Observable
    obs := rxgo.Timer(100*time.Millisecond).SubscribeOn(scheduler)
    
    received := false
    obs.Subscribe(func(item rxgo.Item) {
        received = true
    })
    
    // 推进虚拟时间
    scheduler.AdvanceTimeBy(100 * time.Millisecond)
    
    if !received {
        t.Error("Should have received value")
    }
}

// 获取全局性能统计
stats := rxgo.GetPerformanceStats()
fmt.Printf("Observer调用次数: %d\n", stats.ObserverCalls)
fmt.Printf("内存优化率: %.1f%%\n", stats.MemoryEfficiency)
fmt.Printf("并发Observable数: %d\n", stats.CurrentActiveObs)
fmt.Printf("平均延迟: %.2fms\n", float64(stats.AverageLatency)/1e6)

// 重置统计信息
rxgo.ResetPerformanceStats()

// 检测goroutine泄漏
before := runtime.NumGoroutine()

observable := rxgo.Interval(10*time.Millisecond).Take(100)
subscription := observable.Subscribe(func(item rxgo.Item) {})

subscription.Unsubscribe()
time.Sleep(100*time.Millisecond) // 等待清理

after := runtime.NumGoroutine()
if after > before {
    log.Printf("可能存在goroutine泄漏: %d -> %d", before, after)
}

// ✅ 正确的资源管理
subscription := observable.Subscribe(observer)
defer subscription.Unsubscribe() // 确保资源释放

// ✅ 使用CompositeDisposable管理多个资源
composite := rxgo.NewCompositeDisposable()
composite.Add(subscription1)
composite.Add(subscription2)
defer composite.Dispose()

// ✅ 完整的错误处理
observable.
    Catch(func(err error) rxgo.Observable {
        log.Printf("处理错误: %v", err)
        return rxgo.Just("默认值")
    }).
    Subscribe(onNext, onError, onComplete)

// ✅ 选择合适的调度器
cpuIntensive.SubscribeOn(rxgo.ThreadPoolScheduler)  // CPU密集型
ioOperations.SubscribeOn(rxgo.NewThreadScheduler)   // IO操作
simpleOps.SubscribeOn(rxgo.ImmediateScheduler)      // 简单操作

// ✅ 使用背压控制
flowable.OnBackpressureBuffer().Subscribe(subscriber)

// ✅ 控制并发数量
observable.FlatMap(func(item interface{}) rxgo.Observable {
    return processItem(item)
}, 4) // 最多4个并发处理

// ✅ 使用并行处理大数据集
rxgo.ParallelRange(1, 1000000, runtime.NumCPU()).
    Map(heavyComputation).
    Reduce(combineResults)

type CustomSubject struct {
    *rxgo.PublishSubject
    filter func(interface{}) bool
}

func (cs *CustomSubject) OnNext(value interface{}) {
    if cs.filter(value) {
        cs.PublishSubject.OnNext(value)
    }
}

// 注册全局插件
rxgo.RegisterPlugin("logging", func(source rxgo.Observable) rxgo.Observable {
    return source.DoOnNext(func(value interface{}) {
        log.Printf("流经数据: %v", value)
    })
})

// 使用插件
observable.Transform(rxgo.GetPlugin("logging"))