Build Your Own Redis Pipeline Client in Go

Using a series of eleven articles, the author builds a functional Redis service called EasyRedis , and this part focuses on the pipeline client implementation. The goal is to demonstrate Redis internals rather than provide a superficial overview.

The key insight is that a socket’s send and receive buffers are independent, allowing multiple request packets to be queued in the same connection without waiting for each response. By writing all commands to the send buffer in order, the server processes them sequentially and returns results in the same order, enabling separate goroutines for sending and receiving as long as the order is preserved.

The client code resides in redis/client/client.go and consists of about 200 lines. The RedisClent struct holds the TCP connection, address, status, heartbeat ticker, channels for pending sends and results, and a wait group for in‑flight requests.

Creating the client

type RedisClent struct {
    // socket connection
    conn net.Conn
    addr string
    // client state
    connStatus atomic.Int32
    // heartbeat
    ticker time.Ticker
    // buffer caches
    waitSend   chan *request
    waitResult chan *request
    // tracking active requests
    working sync.WaitGroup
}

func NewRedisClient(addr string) (*RedisClent, error) {
    conn, err := net.Dial("tcp", addr)
    if err != nil {
        return nil, err
    }
    rc := RedisClent{}
    rc.conn = conn
    rc.waitSend = make(chan *request, maxChanSize)
    rc.waitResult = make(chan *request, maxChanSize)
    rc.addr = addr
    return &rc, nil
}

The Start method launches three goroutines: one to continuously pull requests from waitSend and write them to the socket, another to read server replies, and a third to send periodic heartbeats. The connection status is set to running.

Sending Redis commands

Commands are passed as a slice of byte slices ( [][]byte) and placed into rc.waitSend. The method creates a request object, increments counters, pushes it to the send channel, then blocks on a wait group until the response arrives or a timeout occurs.

func (rc *RedisClent) Send(command [][]byte) (protocol.Reply, error) {
    if rc.connStatus.Load() == connClosed {
        return nil, errors.New("client closed")
    }
    req := &request{command: command, wait: wait.Wait{}}
    req.wait.Add(1)
    rc.working.Add(1)
    defer rc.working.Done()
    rc.waitSend <- req
    if req.wait.WaitWithTimeOut(maxWait) {
        return nil, errors.New("time out")
    }
    if req.err != nil {
        return nil, req.err
    }
    return req.reply, nil
}

Sending to the server

The execSend goroutine reads from waitSend and calls sendReq. sendReq attempts to write the request bytes up to three times, retrying only on timeout or deadline errors. On success the request is forwarded to waitResult; on failure the request’s error field is set and its wait group is released.

func (rc *RedisClent) execSend() {
    for req := range rc.waitSend {
        rc.sendReq(req)
    }
}

func (rc *RedisClent) sendReq(req *request) {
    if req == nil || len(req.command) == 0 {
        return
    }
    var err error
    for i := 0; i < 3; i++ {
        _, err = rc.conn.Write(req.Bytes())
        if err == nil || (!strings.Contains(err.Error(), "timeout") && !strings.Contains(err.Error(), "deadline exceeded")) {
            break
        }
    }
    if err == nil {
        rc.waitResult <- req
    } else {
        req.err = err
        req.wait.Done()
    }
}

Receiving from the server

The execReceive goroutine parses the TCP stream using parser.ParseStream. For each payload, if an error indicates a closed connection, the client attempts reconnection; otherwise the reply is handed to handleResult, which matches the reply with the corresponding request from waitResult and signals completion.

func (rc *RedisClent) execReceive() {
    ch := parser.ParseStream(rc.conn)
    for payload := range ch {
        if payload.Err != nil {
            if rc.connStatus.Load() == connClosed {
                return
            }
            rc.reconnect()
            return
        }
        rc.handleResult(payload.Reply)
    }
}

func (rc *RedisClent) handleResult(reply protocol.Reply) {
    req := <-rc.waitResult
    if req == nil {
        return
    }
    req.reply = reply
    req.wait.Done()
}

Reconnection logic

When a network glitch closes the socket, reconnect logs the event, closes the old connection, and attempts to dial the server up to three times. If reconnection fails, the client stops. Otherwise it clears the old waitResult channel, marks pending requests as failed with a "connect reset" error, creates fresh channels, replaces the connection, and restarts the receive goroutine.

func (rc *RedisClent) reconnect() {
    logger.Info("redis client reconnect...")
    rc.conn.Close()
    var conn net.Conn
    for i := 0; i < 3; i++ {
        var err error
        conn, err = net.Dial("tcp", rc.addr)
        if err != nil {
            logger.Error("reconnect error: " + err.Error())
            time.Sleep(time.Second)
            continue
        }
        break
    }
    if conn == nil {
        rc.Stop()
        return
    }
    close(rc.waitResult)
    for req := range rc.waitResult {
        req.err = errors.New("connect reset")
        req.wait.Done()
    }
    rc.waitResult = make(chan *request, maxWait)
    rc.conn = conn
    go rc.execReceive()
}

Overall, the article demonstrates how to decouple sending and receiving via buffered channels, maintain request order, handle timeouts, retries, and automatic reconnection, providing a compact yet complete Redis pipeline client written in Go.