ブラウザで非同期イベントストリームを処理する Observable API

const button = document.getElementById("button");
 
button.when("click").subscribe({
  next: () => alert("clicked via Observable API")
});

const countButton = document.getElementById("count-button");
const endButton = document.getElementById("end-button");
 
// reduce() は Promise を返す
const endCount = await countButton.when("click")
  // endButton がクリックされるまでイベントを監視
  .takeUntil(endButton.when("click"))
  .reduce((count, e) => {
    e.target.textContent = count;
    return count + 1;
}, 0);
 
console.log(endCount);

const container = document.getElementById("container");
 
container.when("click")
  .filter(e => e.target.classList.contains("clickable"))
  .map(e => ({ x: e.clientX, y: e.clientY }))
  .subscribe({
    next: ({ x, y }) => console.log(`clicked at (${x}, ${y})`)
  });

const ws = new WebSocket("ws://localhost:8080");
const messageList = [];
 
ws.when("message")
  .takeUntil(ws.when("close"))
  .takeUntil(ws.when("error"))
  .map(e => JSON.parse(e.data))
  .filter(data => data.type === "message")
  .subscribe({
    next: data => messageList.push(data)
  })
  .finally(() => {
    ws.send(JSON.stringify({ type: "disconnect", userId: "..." }));
  });

const observable = new Observable((subscriber) => {
  let count = 0;
  const id = setInterval(() => {
    // count が 10 以上になったら .complete() を呼び出してイベントストリームを終了
    if (count > 10) {
      subscriber.complete();
      return;
    }
 
    // 10% の確率でエラーを発生
    if (Math.random() < 0.1) {
      subscriber.error(new Error("Something went wrong!"));
      return;
    }
    // 1 秒ごとに count を通知
    subscriber.next(count++);
 
    subscriber.addTeardown(() => {
      console.log("Teardown!");
      clearInterval(id);
    });
  }, 1000);
});
 
observable.subscribe({
  next: (value) => console.log(`Count: ${value}`),
  error: (error) => console.error(error),
  complete: () => console.log("Complete!"),
});

const observable = Observable.from([1, 2, 3, 4, 5])
 
observable.map(value => value * 2)
  .subscribe({
    next: (value) => console.log(value),
    complete: () => console.log("Complete!"),
  });
 
// Output:
// 2
// 4
// 6
// 8
// 10
// Complete!

const abortController = new AbortController();
const signal = abortController.signal;
 
const start = document.getElementById("start");
const stop = document.getElementById("stop");
 
const timer = document.getElementById("timer");
 
start
  .when("click")
  // flatMap() は新しい Observable インスタンスを平坦化して返す
  .flatMap(
    () =>
      new Observable((subscriber) => {
        setInterval(() => {
          subscriber.next(new Date());
        }, 1000);
 
        // `.subscribe()` に渡した `signal` にアクセスできる
        console.log(subscriber.signal)
      }),
  )
  .map((date) => date.toLocaleTimeString())
  .subscribe(
    {
      next: (value) => (timer.textContent = value),
    },
    { signal },
  );
 
// stop ボタンがクリックされたらイベントストリームをキャンセル
stop.when("click").subscribe({
  next: () => abortController.abort(),
});

const input = document.querySelector("input");
const abortController = new AbortController();
const signal = abortController.signal;
 
input
  .when("input")
  .switchMap(
    (e) =>
      new Observable(async (subscriber) => {
        const url = `https://api.github.com/search/repositories?q=${e.target.value}`;
        const res = await fetch(url, { signal: subscriber.signal });
        if (!res.ok) {
          subscriber.error(new Error(`HTTP error! status: ${res.status}`));
          return;
        }
        return subscriber.next(res);
      }),
  )
  .switchMap((response) => response.json())
  // 先頭の 10 件のみ表示
  .map((data) => data.items.slice(0, 10))
  .forEach(
    (items) => {
      // 検索結果を表示する
      // ...  
    },
    { signal },
  )
  // ストリームで発生したエラーをキャッチ
  .catch((e) => console.error(e));

button.when("click")
  .first()
  .then(e => {
    // イベントをキャンセルすることができない
    e.preventDefault();
  });

button.when("click")
  .map(e => {
    e.preventDefault();
    return e;
  })
  .first()
  .then(e => {
    // ...
  });

button.when("click")
  .first(e => {
    e.preventDefault();
    return e;
  })
  .then(e => {
    // ...
  });