Golang桥接模式将多个chan桥接成一个chan

FredricZhu · · 848 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

bridges/bridges.go

// bridges
package bridges

type BridgeO struct {
}

func NewBridge() *BridgeO {
    bridgeO := &BridgeO{}
    return bridgeO
}

func (bridgeO *BridgeO) OrDone(
    done, c <-chan interface{},
) <-chan interface{} {
    valStream := make(chan interface{})
    go func() {
        defer close(valStream)
        for {
            select {
            case <-done:
                return
            case v, ok := <-c:
                if ok == false {
                    return
                }

                select {
                case valStream <- v:
                case <-done:
                }
            }
        }
    }()
    return valStream
}

func (bridgeO *BridgeO) Bridge(
    done <-chan interface{},
    chanStream <-chan <-chan interface{},
) <-chan interface{} {
    valStream := make(chan interface{})
    go func() {
        defer close(valStream)
        for {
            var stream <-chan interface{}
            select {
            case maybeStream, ok := <-chanStream:
                if ok == false {
                    return
                }

                stream = maybeStream
            case <-done:
                return
            }

            for val := range bridgeO.OrDone(done, stream) {
                select {
                case valStream <- val:
                case <-done:
                }
            }
        }
    }()
    return valStream
}

bridge_demo.go

// bridgedemo
package main

import (
    "bridgedemo/bridges"
    "fmt"
)

func main() {
    bridge := bridges.NewBridge()
    genVals := func() <-chan <-chan interface{} {
        chanStream := make(chan (<-chan interface{}))
        go func() {
            defer close(chanStream)
            for i := 0; i < 10; i++ {
                stream := make(chan interface{}, 1)
                stream <- i
                close(stream)
                chanStream <- stream
            }
        }()
        return chanStream
    }

    done := make(chan interface{})
    defer close(done)
    for v := range bridge.Bridge(done, genVals()) {
        fmt.Printf("%v ", v)
    }

    fmt.Println()
}

程序输出如下，