-- 行为：鼠标位置随时间变化
mousePos : Behavior (Int, Int)
mousePos = Mouse.position

-- 事件：鼠标点击
clicks : Event ()
clicks = Mouse.clicks

-- 组合：点击时的鼠标位置
clickPositions : Event (Int, Int)
clickPositions = sampleOn clicks mousePos

-- 转换
countClicks : Behavior Int
countClicks = foldp (\_ acc -> acc + 1) 0 clicks

import { fromEvent, interval, merge, combineLatest } from 'rxjs';
import { map, filter, scan, debounceTime, switchMap, takeUntil } from 'rxjs/operators';

// 基本流
const numbers$ = from([1, 2, 3, 4, 5]);

// 事件流
const clicks$ = fromEvent(document, 'click');
const keys$ = fromEvent(document, 'keydown');

// 变换
const positions$ = clicks$.pipe(
  map(e => ({ x: e.clientX, y: e.clientY }))
);

// 过滤
const enterKeys$ = keys$.pipe(
  filter(e => e.key === 'Enter')
);

// 累积
const clickCount$ = clicks$.pipe(
  scan((count, _) => count + 1, 0)
);

// 防抖
const searchInput$ = fromEvent(searchBox, 'input').pipe(
  debounceTime(300),
  map(e => e.target.value),
  filter(query => query.length > 2)
);

// 搜索请求（取消前一个）
const results$ = searchInput$.pipe(
  switchMap(query => fetch(`/api/search?q=${query}`).then(r => r.json()))
);

// 订阅
results$.subscribe(results => {
  renderResults(results);
});

// 经典应用：搜索自动补全
const searchBox = document.getElementById('search');
const results = document.getElementById('results');

const search$ = fromEvent(searchBox, 'input').pipe(
  // 取输入值
  map(e => e.target.value.trim()),
  // 防抖 300ms
  debounceTime(300),
  // 至少 2 个字符
  filter(query => query.length >= 2),
  // 去重
  distinctUntilChanged(),
  // 发起请求（取消前一个未完成的）
  switchMap(query =>
    from(fetch(`/api/search?q=${encodeURIComponent(query)}`).then(r => r.json())).pipe(
      // 错误处理：返回空结果
      catchError(() => of([]))
    )
  )
);

search$.subscribe(data => {
  results.innerHTML = data.map(item =>
    `<div class="result">${item.title}</div>`
  ).join('');
});

import { merge, combineLatest, zip } from 'rxjs';

// merge：任一流发出值都传播
const merged$ = merge(mouseClicks$, keyPresses$);

// combineLatest：任一流发出时，取其他流的最新值
const fullName$ = combineLatest([firstName$, lastName$]).pipe(
  map(([first, last]) => `${first} ${last}`)
);

// zip：严格配对
const paired$ = zip(names$, ages$).pipe(
  map(([name, age]) => ({ name, age }))
);

import { of, EMPTY } from 'rxjs';
import { catchError, retry, retryWhen, delay } from 'rxjs/operators';

// 基本错误处理
const safeRequest$ = fetch('/api/data').pipe(
  catchError(error => {
    console.error('Request failed:', error);
    return of({ data: [], error: error.message });
  })
);

// 重试
const resilientRequest$ = fetch('/api/data').pipe(
  retry(3)
);

// 指数退避重试
const backoffRequest$ = fetch('/api/data').pipe(
  retryWhen(errors => errors.pipe(
    scan((retryCount, error) => {
      if (retryCount >= 3) throw error;
      return retryCount + 1;
    }, 0),
    delay(retryCount => Math.pow(2, retryCount) * 1000)
  ))
);

// 错误恢复
const requestWithFallback$ = primaryRequest$.pipe(
  catchError(() => fallbackRequest$),
  catchError(() => of(defaultValue))
);

// Redux 本质上是一个流
import { Subject } from 'rxjs';
import { scan, shareReplay } from 'rxjs/operators';

const createStore = (reducer, initialState) => {
  const actions$ = new Subject();
  const state$ = actions$.pipe(
    scan(reducer, initialState),
    shareReplay(1)
  );
  return {
    dispatch: (action) => actions$.next(action),
    select: (selector) => state$.pipe(map(selector)),
    state$,
  };
};

// 使用
const store = createStore(todoReducer, []);

store.select(state => state.filter(t => !t.completed)).subscribe(
  activeTodos => render(activeTodos)
);

store.dispatch({ type: 'ADD_TODO', text: 'Learn FRP' });

// Elm Architecture: Model → Update → View
const ElmApp = (init, update, view) => {
  const actions$ = new Subject();

  const model$ = actions$.pipe(
    scan((model, action) => update(model, action), init)
  );

  const vdom$ = model$.pipe(
    map(model => view(model, (action) => actions$.next(action)))
  );

  return { model$, vdom$ };
};

-- 使用 reflex 库
import Reflex

-- 基本 FRP
example :: Widget t m => m ()
example = do
  -- 创建事件
  rec textInput <- textInput def
      let keypressEvent = _textInput_keypress textInput

  -- 行为：累积按键次数
  count <- foldDyn (+) 0 (1 <$ keypressEvent)

  -- 显示
  display count

-- 动态文本
  dynText $ fmap show count

// 实时数据仪表板
const dashboard$ = combineLatest({
  cpu: interval(1000).pipe(switchMap(() => fetchMetric('cpu'))),
  memory: interval(2000).pipe(switchMap(() => fetchMetric('memory'))),
  requests: fromEventSource('/api/metrics/requests'),
  errors: fromEventSource('/api/metrics/errors').pipe(
    bufferTime(5000),
    filter(errors => errors.length > 0)
  )
}).pipe(
  // 状态转换
  scan((dashboard, metrics) => ({
    ...dashboard,
    ...metrics,
    lastUpdated: Date.now(),
    alerts: [
      ...dashboard.alerts,
      ...(metrics.cpu > 90 ? ['High CPU'] : []),
      ...(metrics.errors.length > 10 ? ['Error spike'] : []),
    ]
  }), { alerts: [] })
);

dashboard$.subscribe(renderDashboard);

const formValidation$ = (form) => {
  const name$ = fromEvent(form.name, 'input').pipe(
    map(e => e.target.value),
    map(name => ({
      value: name,
      valid: name.length >= 2,
      error: name.length < 2 ? 'Name too short' : null
    }))
  );

  const email$ = fromEvent(form.email, 'input').pipe(
    map(e => e.target.value),
    map(email => ({
      value: email,
      valid: /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email),
      error: !/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email) ? 'Invalid email' : null
    }))
  );

  const age$ = fromEvent(form.age, 'input').pipe(
    map(e => parseInt(e.target.value)),
    map(age => ({
      value: age,
      valid: age >= 18 && age <= 120,
      error: age < 18 ? 'Must be 18+' : age > 120 ? 'Invalid age' : null
    }))
  );

  return combineLatest([name$, email$, age$]).pipe(
    map(([name, email, age]) => ({
      fields: { name, email, age },
      valid: name.valid && email.valid && age.valid
    })),
    distinctUntilChanged((a, b) => a.valid === b.valid)
  );
};