Reverse HTTP proxy over WebSocket in Go (Part 4)

app -[1]-> wsp server -[2](WebSocket)-> wsp client -> internal API

func (s *Server) request(w http.ResponseWriter, r *http.Request) {    // (omit): [1]: Receive requests to be proxied    // [2]: Take a WebSocket connection available from pools for relaying received requests.    request := NewConnectionRequest(s.Config.GetTimeout())    // "Dispatcher" is running in a separate thread from the server by `go s.dispatchConnections()`.    // It waits to receive requests to dispatch connection from available pools to clients requests.    // https://github.com/hgsgtk/wsp/blob/ea4902a8e11f820268e52a6245092728efeffd7f/server/server.go#L93    //    // Notify request from handler to dispatcher through Server.dispatcher channel.    s.dispatcher <- request    // Dispatcher tries to find an available connection pool,    // and it returns the connection through Server.connection channel.    // https://github.com/hgsgtk/wsp/blob/ea4902a8e11f820268e52a6245092728efeffd7f/server/server.go#L189    //    // Here waiting for a result from dispatcher.    connection := <-request.connection    if connection == nil {        // It means that dispatcher has set `nil` which is a system error case that is        // not expected in the normal flow.        wsp.ProxyErrorf(w, "Unable to get a proxy connection")        return    }    // [3]: Send the request to the peer through the WebSocket connection.    if err := connection.proxyRequest(w, r); err != nil {        // An error occurred throw the connection away        log.Println(err)        connection.Close()        // Try to return an error to the client        // This might fail if response headers have already been sent        wsp.ProxyError(w, err)    }}

func (s *Server) Register(w http.ResponseWriter, r *http.Request) {    // - 1. Upgrade a received HTTP request to a WebSocket connection    // (omit)    // - 2. Wait a greeting message from the peer and parse it    // (omit)    // 3. Register the connection into server pools.    // (omit)    // Add the WebSocket connection to the pool    pool.Register(ws)}

func (s *Server) Start() {    // (omit)...    // start the "Dispatcher"    go s.dispatchConnections()    // start the "Server"    s.server = &http.Server{        Addr:    s.Config.GetAddr(),        Handler: r,    }    go func() { log.Fatal(s.server.ListenAndServe()) }()}

type Server struct {    // (omit)    // Through dispatcher channel it communicates between "server" thread and "dispatcher" thread.    // "server" thread sends the value to this channel when accepting requests in the endpoint /requests,     // and "dispatcher" thread reads this channel.    dispatcher chan *ConnectionRequest}func (s *Server) dispatchConnections() {    for {        //  The operator <- is "receive operator", which expression blocks until a value is available.        request, ok := <-s.dispatcher        if !ok {            // The value of `ok` is false if it is a zero value generated because the channel is closed an empty.            // In this case, that means server shutdowns.           break        }        for {            // (omit)...            // Verify that we can use this connection            if connection.Take() {                request.connection <- connection                break            }        }        // (omit)...    }}

for {    request, ok := <-s.dispatcher    if !ok {        break    }}

func (s *Server) request(w http.ResponseWriter, r *http.Request) {    // (omit): [1]: Receive requests to be proxied    // Here! Sends a value to dispatcher channel    request := NewConnectionRequest(s.Config.GetTimeout())    s.dispatcher <- request    connection := <-request.connection    if connection == nil {        wsp.ProxyErrorf(w, "Unable to get a proxy connection")        return    }}

type ConnectionRequest struct {    connection chan *Connection}

func (s *Server) request(w http.ResponseWriter, r *http.Request) {    // (omit)    s.dispatcher <- request    // Here!    connection := <-request.connection    if connection == nil {        wsp.ProxyErrorf(w, "Unable to get a proxy connection")        return    }}

type Server struct {    // (omit)    // Through dispatcher channel it communicates between "server" thread and "dispatcher" thread.    // "server" thread sends the value to this channel when accepting requests in the endpoint /requests,     // and "dispatcher" thread reads this channel.    dispatcher chan *ConnectionRequest}func (s *Server) dispatchConnections() {    for {        // Runs in an infinite loop and keeps receiving the value from the `server.dispatcher` channel        // The operator <- is "receive operator", which expression blocks until a value is available.        request, ok := <-s.dispatcher        if !ok {            // The value of `ok` is false if it is a zero value generated because the channel is closed an empty.            // In this case, that means server shutdowns.            break        }        // A timeout is set for each dispatch request.        ctx := context.Background()        ctx, cancel := context.WithTimeout(ctx, s.Config.GetTimeout())        defer cancel()    L:        for {            select {            case <-ctx.Done(): // The timeout elapses                break L            default: // Go through            }            s.lock.RLock()            if len(s.pools) == 0 {                // No connection pool available                s.lock.RUnlock()                break            }            // [1]: Select a pool which has an idle connection            // Build a select statement dynamically to handle an arbitrary number of pools.            cases := make([]reflect.SelectCase, len(s.pools)+1)            for i, ch := range s.pools {                cases[i] = reflect.SelectCase{                    Dir:  reflect.SelectRecv,                    Chan: reflect.ValueOf(ch.idle)}            }            cases[len(cases)-1] = reflect.SelectCase{                Dir: reflect.SelectDefault}            s.lock.RUnlock()            _, value, ok := reflect.Select(cases)            if !ok {                continue // a pool has been removed, try again            }            connection, _ := value.Interface().(*Connection)            // [2]: Verify that we can use this connection and take it.            if connection.Take() {                request.connection <- connection                break            }        }        close(request.connection)    }}

for {    request, ok := <-s.dispatcher    // ...}

// A timeout is set for each dispatch request.ctx := context.Background()ctx, cancel := context.WithTimeout(ctx, s.Config.GetTimeout())defer cancel()L:    for {        select {        case <-ctx.Done(): // The timeout elapses            break L        default: // Go through        }        // ...    }    close(request.connection)

func (s *Server) request(w http.ResponseWriter, r *http.Request) {    // (omit)    s.dispatcher <- request    // Here!    connection := <-request.connection    if connection == nil {        wsp.ProxyErrorf(w, "Unable to get a proxy connection")        return    }}

cases := make([]reflect.SelectCase, len(s.pools)+1)for i, ch := range s.pools {    cases[i] = reflect.SelectCase{        Dir:  reflect.SelectRecv,        Chan: reflect.ValueOf(ch.idle)}}cases[len(cases)-1] = reflect.SelectCase{    Dir: reflect.SelectDefault}s.lock.RUnlock()_, value, ok := reflect.Select(cases)if !ok {    continue // a pool has been removed, try again}connection, _ := value.Interface().(*Connection)

if connection.Take() {    request.connection <- connection    break}