package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    // === バッファなしチャネル（同期型） ===
    unbuffered := make(chan string)
 
    go func() {
        // 受信側が準備できるまでブロックされる
        unbuffered <- "hello"
        fmt.Println("送信完了") // 受信後に出力される
    }()
 
    time.Sleep(100 * time.Millisecond) // 送信側がブロックされていることを確認
    msg := <-unbuffered
    fmt.Println("受信:", msg)
 
    // === バッファ付きチャネル（非同期型） ===
    buffered := make(chan int, 3) // 3要素までバッファリング
 
    // バッファに空きがある限りブロックされない
    buffered <- 10
    buffered <- 20
    buffered <- 30
    // buffered <- 40 // バッファ満杯でブロック（デッドロック）
 
    fmt.Println(<-buffered) // 10（FIFO）
    fmt.Println(<-buffered) // 20
    fmt.Println(<-buffered) // 30
 
    // === チャネルの方向制約 ===
    // chan<- int : 送信専用
    // <-chan int : 受信専用
    producer := func(out chan<- int) {
        for i := 0; i < 5; i++ {
            out <- i
        }
        close(out) // 送信完了を通知
    }
 
    consumer := func(in <-chan int) {
        for v := range in { // close されるまで繰り返す
            fmt.Printf("受信: %d\n", v)
        }
    }
 
    ch := make(chan int, 5)
    go producer(ch)
    consumer(ch) // メインゴルーチンで受信
}

package main
 
import (
    "context"
    "fmt"
    "math/rand"
    "time"
)
 
// タイムアウト付きの処理
func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
    result := make(chan string, 1)
    errCh := make(chan error, 1)
 
    go func() {
        // シミュレーション: ランダムな遅延
        delay := time.Duration(rand.Intn(500)) * time.Millisecond
        time.Sleep(delay)
        result <- fmt.Sprintf("Response from %s (took %v)", url, delay)
    }()
 
    select {
    case res := <-result:
        return res, nil
    case err := <-errCh:
        return "", err
    case <-time.After(timeout):
        return "", fmt.Errorf("timeout after %v", timeout)
    }
}
 
// 最速レスポンスの採用（First Response Wins）
func queryMultipleServers(servers []string) string {
    result := make(chan string, len(servers))
 
    for _, server := range servers {
        go func(s string) {
            // 各サーバーに問い合わせ
            delay := time.Duration(rand.Intn(300)) * time.Millisecond
            time.Sleep(delay)
            result <- fmt.Sprintf("Response from %s", s)
        }(server)
    }
 
    return <-result // 最初のレスポンスを返す
}
 
// Context によるキャンセル伝播
func longRunningTask(ctx context.Context) error {
    for i := 0; ; i++ {
        select {
        case <-ctx.Done():
            fmt.Printf("タスク中断: %v\n", ctx.Err())
            return ctx.Err()
        default:
            fmt.Printf("処理中: ステップ %d\n", i)
            time.Sleep(100 * time.Millisecond)
        }
    }
}
 
func main() {
    // タイムアウトの例
    res, err := fetchWithTimeout("https://example.com", 200*time.Millisecond)
    if err != nil {
        fmt.Println("エラー:", err)
    } else {
        fmt.Println(res)
    }
 
    // 最速レスポンスの例
    servers := []string{"server-a", "server-b", "server-c"}
    fmt.Println(queryMultipleServers(servers))
 
    // Context キャンセルの例
    ctx, cancel := context.WithTimeout(context.Background(), 350*time.Millisecond)
    defer cancel()
    longRunningTask(ctx)
}

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
// Job はワーカーが処理するタスクを表す
type Job struct {
    ID      int
    Payload string
}
 
// Result はジョブの処理結果を表す
type Result struct {
    JobID    int
    Output   string
    Duration time.Duration
}
 
// worker はジョブチャネルから仕事を受け取り、結果チャネルに書き込む
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        start := time.Now()
        // 処理のシミュレーション
        time.Sleep(50 * time.Millisecond)
        output := fmt.Sprintf("Worker %d processed job %d: %s",
            id, job.ID, job.Payload)
        results <- Result{
            JobID:    job.ID,
            Output:   output,
            Duration: time.Since(start),
        }
    }
}
 
func main() {
    const numWorkers = 4
    const numJobs = 20
 
    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
 
    // ワーカーの起動
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }
 
    // ジョブの投入
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Payload: fmt.Sprintf("task-%d", j)}
    }
    close(jobs) // これ以上ジョブがないことを通知
 
    // 全ワーカーの完了を待機し、結果チャネルを閉じる
    go func() {
        wg.Wait()
        close(results)
    }()
 
    // 結果の収集
    for result := range results {
        fmt.Printf("Job %2d: %s (%v)\n",
            result.JobID, result.Output, result.Duration)
    }
}

package main
 
import (
    "fmt"
    "math"
    "sync"
)
 
// generate はスライスの要素をチャネルに送出する（ソースステージ）
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}
 
// square は入力の各値を二乗する（変換ステージ）
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}
 
// filter は条件を満たす値のみ通過させる（フィルタステージ）
func filter(in <-chan int, pred func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if pred(n) {
                out <- n
            }
        }
        close(out)
    }()
    return out
}
 
// fanOut は1つの入力チャネルを n 個のワーカーに分配する
func fanOut(in <-chan int, n int, process func(int) int) []<-chan int {
    outs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        out := make(chan int)
        outs[i] = out
        go func() {
            for v := range in {
                out <- process(v)
            }
            close(out)
        }()
    }
    return outs
}
 
// fanIn は複数のチャネルを1つに統合する
func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}
 
func main() {
    // 基本パイプライン: generate -> square -> filter -> 出力
    isPrime := func(n int) bool {
        if n < 2 {
            return false
        }
        for i := 2; i <= int(math.Sqrt(float64(n))); i++ {
            if n%i == 0 {
                return false
            }
        }
        return true
    }
 
    pipeline := filter(
        square(generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)),
        isPrime,
    )
 
    fmt.Println("素数である平方数:")
    for v := range pipeline {
        fmt.Println(v) // 4, 9, 25, 49 が出力される
    }
}

use std::sync::mpsc;
use std::thread;
use std::time::Duration;
 
fn main() {
    // === 基本的な使用 ===
    let (tx, rx) = mpsc::channel();
 
    thread::spawn(move || {
        let data = String::from("Hello from thread!");
        tx.send(data).unwrap();
        // data は move されたため、ここでは使用不可
        // println!("{}", data); // コンパイルエラー!
    });
 
    let received = rx.recv().unwrap();
    println!("受信: {}", received);
 
    // === 複数プロデューサー ===
    let (tx, rx) = mpsc::channel();
 
    for i in 0..5 {
        let tx_clone = tx.clone();
        thread::spawn(move || {
            let msg = format!("スレッド {} からのメッセージ", i);
            tx_clone.send(msg).unwrap();
            thread::sleep(Duration::from_millis(100));
        });
    }
    drop(tx); // 元の tx を破棄（全送信者が drop されると rx が終了）
 
    // イテレータとして受信（全送信者が drop されるまで）
    for msg in rx {
        println!("{}", msg);
    }
 
    // === 同期チャネル（バッファ付き） ===
    let (tx, rx) = mpsc::sync_channel(2); // バッファサイズ 2
 
    thread::spawn(move || {
        tx.send(1).unwrap(); // 即座に返る
        tx.send(2).unwrap(); // 即座に返る
        println!("バッファ満杯前");
        tx.send(3).unwrap(); // バッファ満杯、受信されるまでブロック
        println!("3つ目送信完了");
    });
 
    thread::sleep(Duration::from_millis(500));
    println!("受信開始");
    for val in rx {
        println!("値: {}", val);
    }
}

use crossbeam_channel::{bounded, unbounded, select, Receiver, Sender};
use std::thread;
use std::time::Duration;
 
// タイムアウト付きのリクエスト処理
fn request_with_timeout(timeout: Duration) -> Result<String, String> {
    let (tx, rx) = bounded(1);
 
    thread::spawn(move || {
        // 処理のシミュレーション
        thread::sleep(Duration::from_millis(200));
        let _ = tx.send("処理完了".to_string());
    });
 
    select! {
        recv(rx) -> msg => msg.map_err(|e| e.to_string()),
        default(timeout) => Err("タイムアウト".to_string()),
    }
}
 
// ファンイン: 複数のソースを1つに統合
fn fan_in(receivers: Vec<Receiver<String>>) -> Receiver<String> {
    let (tx, rx) = unbounded();
 
    for r in receivers {
        let tx = tx.clone();
        thread::spawn(move || {
            for msg in r {
                if tx.send(msg).is_err() {
                    break;
                }
            }
        });
    }
 
    rx
}
 
fn main() {
    // タイムアウトの例
    match request_with_timeout(Duration::from_millis(300)) {
        Ok(msg) => println!("成功: {}", msg),
        Err(e) => println!("失敗: {}", e),
    }
 
    // 複数チャネルの select
    let (s1, r1) = bounded::<String>(0);
    let (s2, r2) = bounded::<String>(0);
 
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(100));
        s1.send("チャネル1".to_string()).unwrap();
    });
 
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(200));
        s2.send("チャネル2".to_string()).unwrap();
    });
 
    // 両方のチャネルを待機
    for _ in 0..2 {
        select! {
            recv(r1) -> msg => println!("r1: {:?}", msg),
            recv(r2) -> msg => println!("r2: {:?}", msg),
        }
    }
}

# === GenServer を使ったアクター（Elixir） ===
defmodule BankAccount do
  use GenServer
 
  # --- クライアント API ---
  def start_link(initial_balance) do
    GenServer.start_link(__MODULE__, initial_balance)
  end
 
  def deposit(account, amount) when amount > 0 do
    GenServer.call(account, {:deposit, amount})
  end
 
  def withdraw(account, amount) when amount > 0 do
    GenServer.call(account, {:withdraw, amount})
  end
 
  def balance(account) do
    GenServer.call(account, :balance)
  end
 
  # 非同期通知（返答不要）
  def notify(account, message) do
    GenServer.cast(account, {:notify, message})
  end
 
  # --- サーバーコールバック ---
  @impl true
  def init(initial_balance) do
    {:ok, %{balance: initial_balance, history: []}}
  end
 
  @impl true
  def handle_call({:deposit, amount}, _from, state) do
    new_balance = state.balance + amount
    new_state = %{
      balance: new_balance,
      history: [{:deposit, amount, new_balance} | state.history]
    }
    {:reply, {:ok, new_balance}, new_state}
  end
 
  @impl true
  def handle_call({:withdraw, amount}, _from, state) do
    if state.balance >= amount do
      new_balance = state.balance - amount
      new_state = %{
        balance: new_balance,
        history: [{:withdraw, amount, new_balance} | state.history]
      }
      {:reply, {:ok, new_balance}, new_state}
    else
      {:reply, {:error, :insufficient_funds}, state}
    end
  end
 
  @impl true
  def handle_call(:balance, _from, state) do
    {:reply, state.balance, state}
  end
 
  @impl true
  def handle_cast({:notify, message}, state) do
    IO.puts("通知: #{message}")
    {:noreply, state}
  end
end
 
# 使用例
{:ok, account} = BankAccount.start_link(1000)
{:ok, balance} = BankAccount.deposit(account, 500)
IO.puts("残高: #{balance}")  # => 残高: 1500
 
{:ok, balance} = BankAccount.withdraw(account, 200)
IO.puts("残高: #{balance}")  # => 残高: 1300
 
{:error, :insufficient_funds} = BankAccount.withdraw(account, 5000)
IO.puts("残高: #{BankAccount.balance(account)}")  # => 残高: 1300

# 監視ツリーの定義
defmodule MyApp.Supervisor do
  use Supervisor
 
  def start_link(opts) do
    Supervisor.start_link(__MODULE__, :ok, opts)
  end
 
  @impl true
  def init(:ok) do
    children = [
      # 子プロセスの仕様
      {BankAccount, 0},
      {MyApp.Cache, []},
      {MyApp.EventBus, []}
    ]
 
    # one_for_one: 個別に再起動
    # max_restarts: 5秒間に3回まで再起動を許容
    Supervisor.init(children,
      strategy: :one_for_one,
      max_restarts: 3,
      max_seconds: 5
    )
  end
end

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
// EventBus はトピックベースの Pub/Sub を実装する
type EventBus struct {
    mu          sync.RWMutex
    subscribers map[string][]chan interface{}
}
 
func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]chan interface{}),
    }
}
 
// Subscribe はトピックを購読し、メッセージを受信するチャネルを返す
func (eb *EventBus) Subscribe(topic string, bufferSize int) <-chan interface{} {
    eb.mu.Lock()
    defer eb.mu.Unlock()
 
    ch := make(chan interface{}, bufferSize)
    eb.subscribers[topic] = append(eb.subscribers[topic], ch)
    return ch
}
 
// Publish はトピックの全購読者にメッセージを送信する
func (eb *EventBus) Publish(topic string, msg interface{}) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
 
    for _, ch := range eb.subscribers[topic] {
        // 非ブロッキング送信（バッファ満杯なら破棄）
        select {
        case ch <- msg:
        default:
            // 購読者が追いつかない場合は破棄
            fmt.Printf("警告: トピック %s の購読者がメッセージを処理できません\n", topic)
        }
    }
}
 
// Close はトピックの全チャネルを閉じる
func (eb *EventBus) Close(topic string) {
    eb.mu.Lock()
    defer eb.mu.Unlock()
 
    for _, ch := range eb.subscribers[topic] {
        close(ch)
    }
    delete(eb.subscribers, topic)
}
 
func main() {
    bus := NewEventBus()
 
    // 購読者 A: "orders" トピック
    ordersA := bus.Subscribe("orders", 10)
    go func() {
        for msg := range ordersA {
            fmt.Printf("[購読者A] 注文受信: %v\n", msg)
        }
    }()
 
    // 購読者 B: "orders" トピック
    ordersB := bus.Subscribe("orders", 10)
    go func() {
        for msg := range ordersB {
            fmt.Printf("[購読者B] 注文受信: %v\n", msg)
        }
    }()
 
    // 購読者 C: "logs" トピック
    logs := bus.Subscribe("logs", 100)
    go func() {
        for msg := range logs {
            fmt.Printf("[ログ] %v\n", msg)
        }
    }()
 
    // 発行
    bus.Publish("orders", map[string]interface{}{
        "id": 1, "item": "Book", "price": 2980,
    })
    bus.Publish("orders", map[string]interface{}{
        "id": 2, "item": "Pen", "price": 150,
    })
    bus.Publish("logs", "Order processing started")
 
    time.Sleep(100 * time.Millisecond)
    bus.Close("orders")
    bus.Close("logs")
}

package main
 
import (
    "fmt"
    "time"
)
 
// Request は応答チャネルを含むリクエスト
type Request struct {
    Query    string
    ReplyCh  chan Response  // 応答を受け取るチャネル
}
 
// Response はリクエストへの応答
type Response struct {
    Answer string
    Err    error
}
 
// server はリクエストを受け取って処理するゴルーチン
func server(requests <-chan Request) {
    for req := range requests {
        // リクエストの処理
        time.Sleep(50 * time.Millisecond) // 処理時間
        req.ReplyCh <- Response{
            Answer: fmt.Sprintf("Answer to: %s", req.Query),
            Err:    nil,
        }
    }
}
 
// ask はサーバーにリクエストを送り、応答を待つ
func ask(requests chan<- Request, query string, timeout time.Duration) (Response, error) {
    replyCh := make(chan Response, 1)
    requests <- Request{Query: query, ReplyCh: replyCh}
 
    select {
    case resp := <-replyCh:
        return resp, resp.Err
    case <-time.After(timeout):
        return Response{}, fmt.Errorf("request timed out")
    }
}
 
func main() {
    requests := make(chan Request, 10)
    go server(requests)
 
    resp, err := ask(requests, "What is message passing?", time.Second)
    if err != nil {
        fmt.Println("エラー:", err)
    } else {
        fmt.Println("応答:", resp.Answer)
    }
 
    close(requests)
}

// === アンチパターン: チャネルによるカウンター ===
// 不必要に複雑で、パフォーマンスも悪い
type ChannelCounter struct {
    incrementCh chan struct{}
    getCh       chan chan int
}
 
func NewChannelCounter() *ChannelCounter {
    c := &ChannelCounter{
        incrementCh: make(chan struct{}),
        getCh:       make(chan chan int),
    }
    go func() {
        count := 0
        for {
            select {
            case <-c.incrementCh:
                count++
            case replyCh := <-c.getCh:
                replyCh <- count
            }
        }
    }()
    return c
}
 
func (c *ChannelCounter) Increment() { c.incrementCh <- struct{}{} }
func (c *ChannelCounter) Get() int {
    replyCh := make(chan int, 1)
    c.getCh <- replyCh
    return <-replyCh
}
 
// === 推奨: sync/atomic による単純なカウンター ===
// シンプルで高速
import "sync/atomic"
 
type AtomicCounter struct {
    count int64
}
 
func (c *AtomicCounter) Increment() { atomic.AddInt64(&c.count, 1) }
func (c *AtomicCounter) Get() int64 { return atomic.LoadInt64(&c.count) }
 
// 判断基準:
// - 単純な状態の保護 → sync.Mutex / sync/atomic
// - データの所有権移動 → チャネル
// - 複数の非同期イベントの待機 → チャネル + select
// - パイプライン処理 → チャネル

// === アンチパターン: ゴルーチンリーク ===
func leakySearch(query string) string {
    results := make(chan string)
 
    // 3つのバックエンドに問い合わせ
    go func() { results <- searchBackendA(query) }()
    go func() { results <- searchBackendB(query) }()
    go func() { results <- searchBackendC(query) }()
 
    // 最初の結果だけ返す
    return <-results
    // 問題: 残り2つのゴルーチンは永遠にブロックされる!
    // results に送信しようとするが、受信者がいない
}
 
// === 修正版: Context によるキャンセルとバッファ ===
func safeSearch(ctx context.Context, query string) string {
    // バッファサイズを送信者数と同じにする
    results := make(chan string, 3)
 
    ctx, cancel := context.WithCancel(ctx)
    defer cancel() // 最初の結果を得たら他をキャンセル
 
    search := func(backend func(context.Context, string) string) {
        select {
        case results <- backend(ctx, query):
        case <-ctx.Done():
            return // キャンセルされたら終了
        }
    }
 
    go search(searchBackendA)
    go search(searchBackendB)
    go search(searchBackendC)
 
    return <-results
}
 
// キャンセルを受け取るバックエンドの例
func searchBackendA(ctx context.Context, query string) string {
    select {
    case <-time.After(200 * time.Millisecond):
        return "Result from A"
    case <-ctx.Done():
        return "" // キャンセル時は即座に戻る
    }
}

// === アンチパターン: チャネル間のデッドロック ===
func deadlock() {
    chA := make(chan int)
    chB := make(chan int)
 
    // ゴルーチン 1: chA に送信してから chB を受信
    go func() {
        chA <- 1      // chB が受信するまでブロック
        val := <-chB  // 永遠に到達しない
        fmt.Println(val)
    }()
 
    // ゴルーチン 2: chB に送信してから chA を受信
    go func() {
        chB <- 2      // chA が受信するまでブロック
        val := <-chA  // 永遠に到達しない
        fmt.Println(val)
    }()
 
    // 両方のゴルーチンが互いを待ってデッドロック!
}
 
// === 修正版: select で非決定的に選択 ===
func noDeadlock() {
    chA := make(chan int, 1)
    chB := make(chan int, 1)
 
    go func() {
        for i := 0; i < 5; i++ {
            select {
            case chA <- i:
            case val := <-chB:
                fmt.Println("G1 received:", val)
            }
        }
    }()
 
    go func() {
        for i := 0; i < 5; i++ {
            select {
            case chB <- i + 100:
            case val := <-chA:
                fmt.Println("G2 received:", val)
            }
        }
    }()
 
    time.Sleep(time.Second)
}

// === アンチパターン: 無制限バッファ（メモリ枯渇の危険）===
func unboundedBuffer() {
    // Go のチャネルは固定バッファだが、
    // スライスで無制限キューを実装するとメモリ枯渇のリスク
    type UnboundedChan struct {
        in   chan int
        out  chan int
        buf  []int
    }
    // 生産速度 > 消費速度の場合、buf が無限に成長する
 
    // === 推奨: バックプレッシャーの設計 ===
    // バッファ付きチャネルで自然なバックプレッシャー
    ch := make(chan int, 100) // 100 要素で生産者がブロック
 
    // または明示的なレート制限
    // rate.NewLimiter(rate.Every(time.Millisecond), 100)
}