"""
对比 vools-reactive 与 Rx 完整规范的功能差距
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def get_vools_features():
"""获取 vools-reactive 已实现的功能"""
from vools.reactive import Observable, ops
from vools.reactive.subject import Subject, BehaviorSubject, ReplaySubject, AsyncSubject
return {
'creating': {
'Create': False,
'Empty': True,
'Never': True,
'Throw': True,
'From': True,
'FromIterable': True,
'Interval': True,
'Just': True,
'Of': True,
'Timer': False,
'Defer': False,
'FromCallable': False,
'FromFuture': False,
'Range': False,
'Repeat': False,
'Start': False,
},
'transforming': {
'Buffer': True,
'FlatMap': True,
'GroupBy': True,
'Map': True,
'Scan': True,
'Window': False,
'ConcatMap': True,
'FlatMapLatest': True,
},
'filtering': {
'Debounce': True,
'Distinct': False,
'Filter': True,
'First': False,
'Last': False,
'Skip': True,
'Take': True,
'ElementAt': False,
'IgnoreElements': False,
'Sample': False,
'SkipLast': False,
'TakeLast': False,
'ThrottleFirst': True,
'ThrottleLatest': False,
'DistinctUntilChanged': True,
},
'combining': {
'CombineLatest': True,
'Concat': True,
'Merge': True,
'StartWith': True,
'Zip': True,
'Amb': False,
'Switch': False,
'And': False,
'Join': False,
'WithLatestFrom': True,
'EndWith': True,
},
'error_handling': {
'Catch': True,
'Retry': True,
'Finally': False,
'OnErrorResumeNext': True,
'OnErrorReturn': True,
'RetryWhen': True,
},
'utility': {
'Delay': True,
'Do': True,
'DoOnCompleted': False,
'DoOnError': False,
'DoOnNext': False,
'ObserveOn': False,
'Subscribe': True,
'SubscribeOn': False,
'Tap': True,
'Timeout': False,
'Dematerialize': False,
'Materialize': False,
'TimeInterval': False,
'Timestamp': False,
'Using': False,
'Serialize': False,
},
'conditional': {
'All': True,
'Every': True,
'SkipUntil': False,
'SkipWhile': True,
'TakeUntil': True,
'TakeWhile': True,
'Amb': False,
'Contains': True,
'DefaultIfEmpty': False,
'IsEmpty': True,
'SequenceEqual': False,
'Any': True,
},
'mathematical': {
'Average': True,
'Count': True,
'Max': True,
'Min': True,
'Reduce': True,
'Sum': True,
'ToList': True,
'ToMap': False,
'ToSet': False,
},
'connectable': {
'Connect': False,
'Publish': False,
'PublishReplay': False,
'RefCount': False,
'Replay': True,
'Share': False,
},
'backpressure': {
'BackpressureBuffer': False,
'BackpressureDrop': False,
'BackpressureError': False,
'BackpressureLatest': False,
},
'subjects': {
'Subject': True,
'BehaviorSubject': True,
'ReplaySubject': True,
'AsyncSubject': True,
'PublishSubject': True,
},
'schedulers': {
'Immediate': False,
'CurrentThread': False,
'AsyncIO': False,
'ThreadPool': False,
'NewThread': False,
}
}
def analyze_gaps():
"""分析功能差距"""
vools = get_vools_features()
print("=" * 80)
print("vools-reactive vs Rx 完整规范功能对比")
print("=" * 80)
total_ops = 0
total_implemented = 0
category_stats = []
for category, ops in vools.items():
category_total = len(ops)
category_impl = sum(1 for v in ops.values() if v)
category_pct = (category_impl / category_total) * 100
total_ops += category_total
total_implemented += category_impl
category_stats.append({
'name': category.replace('_', ' ').title(),
'total': category_total,
'impl': category_impl,
'pct': category_pct,
'ops': ops
})
print("\n📊 总体统计:")
print(f" 总操作符数: {total_ops}")
print(f" 已实现: {total_implemented}")
print(f" 覆盖率: {(total_implemented / total_ops) * 100:.1f}%")
print("\n📋 分类统计:")
print("-" * 60)
print(f"{'类别':<20} {'总数':>6} {'已实现':>6} {'覆盖率':>10}")
print("-" * 60)
for stat in category_stats:
print(f"{stat['name']:<20} {stat['total']:>6} {stat['impl']:>6} {stat['pct']:>9.1f}%")
print("\n🔍 缺失功能明细 (按优先级排序):")
print("-" * 80)
high_priority = [
('creating', ['Timer', 'Defer', 'Repeat']),
('filtering', ['First', 'Last', 'Distinct', 'ElementAt']),
('combining', ['Amb', 'Switch', 'Join']),
('utility', ['ObserveOn', 'SubscribeOn', 'Timeout', 'Timestamp']),
('conditional', ['SkipUntil', 'DefaultIfEmpty', 'SequenceEqual']),
('mathematical', ['ToMap', 'ToSet']),
('connectable', ['Connect', 'Publish', 'PublishReplay', 'RefCount', 'Share']),
('backpressure', ['BackpressureBuffer', 'BackpressureDrop', 'BackpressureLatest']),
('schedulers', ['Immediate', 'CurrentThread', 'AsyncIO', 'ThreadPool', 'NewThread']),
]
for category, missing_ops in high_priority:
ops_dict = vools[category]
actual_missing = [op for op in missing_ops if not ops_dict.get(op, False)]
if actual_missing:
category_name = category.replace('_', ' ').title()
print(f"\n{category_name}:")
print(f" ❌ {', '.join(sorted(actual_missing))}")
print("\n✨ vools-reactive 独有功能:")
print("-" * 80)
unique_features = [
"• placeholder 表达式支持: ops.filter('_ > 0')",
"• curry 柯里化集成: ops.map(curry(fn)(arg))",
"• >> 管道操作符: obs >> ops.filter(cond) >> ops.map(fn)",
"• lazy 模块集成: 延迟执行支持",
"• Subscription 上下文管理器: with obs.subscribe(...) as sub:",
"• debug.trace 装饰器: 函数调用追踪",
"• combine_latest 性能: 比 RxPy 快 5.77x",
]
for feature in unique_features:
print(feature)
print("\n🎯 建议下一步实现:")
print("-" * 80)
suggestions = [
("高优先级", ["Timer", "Defer", "Repeat", "First", "Last", "ObserveOn", "SubscribeOn"]),
("中优先级", ["Distinct", "ElementAt", "Amb", "Switch", "Timeout", "Timestamp"]),
("低优先级", ["Backpressure*", "Connectable*", "Window", "DoOn*"]),
]
for priority, features in suggestions:
print(f"\n{priority}:")
print(f" {' '.join(sorted(features))}")
if __name__ == "__main__":
analyze_gaps()
