RGC 网络编程入门:TCP、UDP 与 HTTP 客户端

RGC 标准库提供了 TCP / UDP / HTTP/1.1 / HTTP/2 / HTTPS 网络能力，既有同步 API，也提供非阻塞原语与 epoll 事件循环；全部以源码模块形式放在 lib/ 下，用 RGC 自身封装：

模块	用途
net	同步 TCP（连接/监听/收发/超时/Unix 域套接字）
udp	数据报（发送/接收/IPv6/二进制/超时）
http	HTTP 客户端（含 HTTPS、HTTP/2 自动协商、重定向、gzip、Cookie、代理、连接池）
http2	低层 HTTP/2（ALPN、HPACK、流控、trailer、帧编解码、get_many 并发多路复用）
https	HTTPS 兼容层（http 已原生支持 https://）
tls	低层 TLS 连接（自定义收发协议、mTLS 客户端证书、ALPN）
bytes	二进制安全的字节缓冲（网络收发辅助）
event	epoll 事件循环（fd 注册/等待、timerfd 定时器、跨线程唤醒）

底层通过 extern def 直接调用 libc socket；TLS 与 gzip 已是纯 RGC 实现（无 OpenSSL/zlib 依赖），链接时不需要额外配置链接参数。

一个贯穿全篇的设计：网络操作的结果用 ok: bool + error: str 的包装类返回，失败不 panic、不抛异常。连接被拒绝、解析失败、端口越界都会得到一个 ok=False 的结果，你负责检查并处理。

1. 准备

编译器会自动在标准库目录（lib/）查找模块，直接编译运行即可：

rgc main.rg -o prog
./prog

本篇所有示例都在本仓库用 src/build/rgc3 编译运行验证过，可直接复制。

2. TCP：连接与监听

2.1 一个完整的 TCP 回环示例

下面这个程序单进程内完成「监听 → 连接 → 接受 → 双向收发 → 关闭」全流程。listen_on 使用端口 0 时，内核会分配一个临时端口：

import net

def main() -> int {
    # 1) 监听：端口 0 让内核分配临时端口
    l = net.listen_on("127.0.0.1", 0)
    if not l.ok {
        print("listen failed:", l.error)
        return 1
    }
    listener = l.listener
    port = listener.local_port()
    print("listening on port", port)

    # 2) 客户端连接（backlog 让连接先排队）
    c = net.connect_to("127.0.0.1", port)
    if not c.ok {
        print("connect failed:", c.error)
        listener.close()
        return 1
    }
    client = c.stream

    # 3) 服务器接受连接
    a = listener.accept()
    if not a.ok {
        print("accept failed:", a.error)
        client.close()
        listener.close()
        return 1
    }
    server = a.stream

    # 4) 双向收发
    client.send("ping")
    r1 = server.recv(64)
    print("server got:", r1.value)     # ping

    server.send("pong")
    r2 = client.recv(64)
    print("client got:", r2.value)     # pong

    # 5) 关闭：对端会读到 EOF（ok=True, value=""）
    client.close()
    eof = server.recv(16)
    print("eof:", eof.ok, eof.value.len())   # True 0

    server.close()
    listener.close()
    return 0
}

运行输出：

listening on port 43727
server got: ping
client got: pong
eof: True 0

几个要点：

• send 阻塞直到全部字节写出，返回实际发送字节数；
• recv(max) 最多读取 max 字节；对端关闭（EOF）时返回 ok=True, value=""；
• 真实错误（连接拒绝、解析失败、超时）返回 ok=False + error 文本；
• close() 重复调用安全，不会崩溃。

2.2 TCP API 一览

API	签名	说明
连接	net.connect(addr: str) -> TcpConnect	地址形如 "host:port"，按最后一个 : 切分
连接	net.connect_to(host: str, port: int) -> TcpConnect	结构化形式，端口 1–65535
超时	net.connect_timeout(addr, ms) / connect_to_timeout(host, port, ms)	连接超时（毫秒，0 阻塞）
监听	net.listen(addr) / net.listen_on(host, port)	SO_REUSEADDR，backlog 128；端口 0 为临时端口
解析	net.resolve(host) -> List[str]	getaddrinfo 返回的地址列表（IPv4/IPv6）
发送	TcpStream.send(data: str) -> int	阻塞写完全部字节
发送	TcpStream.send_bytes(data: @Bytes) -> int	二进制安全（显式长度）
接收	TcpStream.recv(max) -> TcpRecv	最多 max 字节；EOF 返回空串
接收	TcpStream.recv_bytes(max) -> TcpRecvBytes	二进制安全
超时	TcpStream.set_timeout(ms)	设置后续收发超时；0 恢复阻塞
半关闭	TcpStream.shutdown_write()	发送 FIN，仍可接收
地址	TcpStream.local_port() / peer_host() / peer_port()	本端/对端地址信息
接受	TcpListener.accept() / accept_timeout(ms)	阻塞等待连接
端口	TcpListener.local_port() -> int	实际监听端口
Unix	net.connect_unix(path) / net.listen_unix(path)	Unix 域套接字，@name 表示抽象地址空间

结果包装类：TcpConnect / TcpListen / TcpAccept / TcpRecv，都带 ok: bool 与 error: str 字段（失败时 stream.fd == -1）。

3. UDP：数据报

UDP 是无连接的，udp.open 绑定一个本地端口，send_to / recv_from 每次收发一个数据报：

import bytes
import udp

def main() -> int {
    a = udp.open("127.0.0.1:0")
    b = udp.open("127.0.0.1:0")
    if not a.ok or not b.ok {
        print("bind failed")
        return 1
    }
    s1 = a.sock
    s2 = b.sock
    port2 = s2.local_port()

    # 发送数据报
    r = s1.send_to("ping", "127.0.0.1", port2)
    print("sent:", r.ok, r.sent)                     # True 4
    r2 = s2.recv_from(64)
    print("got:", r2.data, r2.host, r2.port == s1.local_port())

    # 数据报边界：两次发送 -> 两次接收
    s1.send_to("one", "127.0.0.1", port2)
    s1.send_to("two", "127.0.0.1", port2)
    print("dgram:", s2.recv_from(64).data, s2.recv_from(64).data)   # one two

    # 二进制安全：data_bytes 保留全部字节
    payload = bytes.bytes_new(2)
    payload.push(0)
    payload.push(255)
    s1.send_bytes(payload, "127.0.0.1", port2)
    r3 = s2.recv_from(16)
    print("binary:", r3.data_bytes.len, r3.data_bytes.get(0), r3.data_bytes.get(1))

    # 读超时
    s2.set_timeout(100)
    r4 = s2.recv_from(16)
    print("timeout:", r4.ok, r4.error)              # False timeout

    s1.close()
    s2.close()
    return 0
}

运行输出：

sent: True 4
got: ping 127.0.0.1 True
dgram: one two
binary: 2 0 255
timeout: False timeout

要点：

• 一次 recv_from 对应一个数据报，不会与相邻数据报拼接；
• data 按 C 字符串处理（遇 NUL 截断）；需要二进制数据用 data_bytes 和 send_bytes；
• set_timeout(ms) 后超时返回 ok=False, error="timeout"，0 恢复阻塞；
• IPv6 直接 udp.open_on("::1", 0)；主机名自动解析（localhost 等）。

4. HTTP 客户端

4.1 一次性请求

import http 提供模块级函数，一次调用完成请求：

import http

def main() {
    r = http.get("http://127.0.0.1:8080/hello")
    if not r.ok {
        print("request failed:", r.error)          # 网络/证书/限额错误
        return
    }
    res = r.response
    print("status:", res.status)                   # 200
    print("type:", http.header(@res, "Content-Type"))
    print("body:", res.body)
}

• http.get(url) / http.post(url, body) / http.request(method, url, body)；
• 响应结构 HttpResponse：status、version、headers、body、body_bytes、url（重定向后的最终地址）；
• http.header(@res, "Content-Type") 大小写不敏感地查响应头，未找到返回 ""；
• body 是已解压的文本（strlen 语义），二进制内容用 body_bytes；
• 默认自动跟随 3xx 重定向，自动解压 gzip / deflate。

4.2 复用连接的客户端

需要 keep-alive、Cookie、自定义头、代理或限额时，使用 http.new_client()：

import http

def main() {
    c = http.new_client()
    c.add_header("X-Token", "secret")          # 每次请求附加
    c.timeout_ms = 5000                          # 读超时
    c.max_body = 1 << 20                        # 正文上限 1 MiB

    a = c.get("http://127.0.0.1:8080/a")       # 同一 keep-alive 连接
    b = c.post("http://127.0.0.1:8080/b", "data")
    print(a.response.status, b.response.status)
    c.close()
}

常用配置字段：

字段	默认	说明
timeout_ms	30000	单次请求读超时
connect_timeout_ms	10000	TCP/TLS 连接与握手超时
max_redirects	10	超过返回 too many redirects
max_body	32 MiB	解压前/后的正文上限
max_header	64 KiB	单个响应头上限
proxy	""	HTTP 代理（HTTPS 走 CONNECT 隧道）
verify_tls	True	是否校验证书链与主机名
keep_alive	True	是否复用连接
follow_redirects	True	是否自动跟随 3xx
compress_request	False	请求体 gzip 压缩
headers / cookies	[]	附加头 / Cookie 罐

4.3 查询串编解码

pairs = List[str]()
pairs.push("q=a b")
pairs.push("x=1")
print(http.encode_query(pairs))        # q=a%20b&x=1

dq = http.decode_query("q=hello+world&x=%E4%BD%A0")
print(dq.get(0), dq.get(1))             # q=hello world x=你

4.4 异步请求

async def 函数在调用点派生线程，返回 Task[T]，配合内建 asyncio 汇合：

import asyncio
import http

def main() -> int {
    tasks = List[Task[HttpResult]]()
    tasks.push(http.get_async("http://127.0.0.1:8080/a"))
    tasks.push(http.get_async("http://127.0.0.1:8080/b"))
    rs = asyncio.run(asyncio.gather(tasks))
    print(rs.len(), rs.get(0).ok, rs.get(1).ok)
    return 0
}

4.5 事件循环与非阻塞 IO

需要同时等待大量 fd 时，event 模块提供 epoll 事件循环（定时器基于 timerfd， 可跨线程唤醒）。配合 net / udp 的非阻塞原语，单个线程即可管理大量连接：

import event
import net

def main() {
    loop = event.new_loop()
    l = net.listen_on("127.0.0.1", 8080)
    listener = l.listener
    listener.set_nonblocking(True)
    loop.add(listener.fd, 1, event.ev_in())

    while True {
        w = loop.wait(-1)                  # -1 无限等待，0 只轮询一次
        i = 0
        while i < w.events.len() {
            e = w.events.get(i)
            if e.token == 1 {
                a = listener.try_accept()
                if a.ok and not a.again {
                    # a.stream 用 try_recv / try_send 处理
                }
            }
            i = i + 1
        }
    }
}

connect_start / connect_finish 做非阻塞连接；try_accept / try_recv / try_send 在 again=True 时表示当前无数据或不可写（EAGAIN），应等待下一次 事件。add_timer / add_interval 提供单次/周期定时器，loop.post(token) 与 event.post_fd(loop.wake_w, token) 可唤醒等待线程。事件循环与线程池版 async 相互独立，await 尚未接入。

5. HTTPS 与 TLS

http 模块原生支持 https://（默认校验证书链与主机名），无需额外配置。需要跳过证书校验（本地自签名服务）时用 https 兼容层：

import https

def main() {
    r = https.get("https://example.com/")
    if r.ok {
        print(r.response.status)
    } else {
        print("request failed:", r.error)      # 证书/网络错误
    }

    local = https.get_insecure("https://127.0.0.1:8443/")
    print(local.ok)                              # 跳过证书校验，仅供本地测试
}

需要低层 TLS 时用 tls.connect：

import tls

def main() {
    c = tls.connect("example.com", 443, 10000, True)   # 超时 10s，校验证书
    if c.ok {
        s = c.conn
        s.send("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
        print(s.recv(4096).value)
        s.close()
    }
}

6. 实战：自包含的 HTTP 请求工具

现在把 net 和 http 结合起来，写一个自包含的程序：main 在后台用 --serve 参数启动自己作为最小 HTTP 服务器，然后作为客户端用 http.get / http.post 请求它。这样不需要任何外部服务，单文件即可跑通。

import http
import io
import net
import os
import time

# --serve 模式：最小 HTTP 服务器（响应 3 个连接）
def serve(port_file: str) -> int {
    l = net.listen_on("127.0.0.1", 0)
    if not l.ok {
        return 1
    }
    listener = l.listener
    port = listener.local_port()
    wr = io.write_file(port_file, to_string(port))
    if not wr.is_ok() {
        listener.close()
        return 1
    }

    served = 0
    while served < 3 {
        a = listener.accept()
        if not a.ok {
            break
        }
        s = a.stream
        buf = ""
        while not buf.contains("\r\n\r\n") {
            r = s.recv(256)
            if not r.ok or r.value.len() == 0 {
                break
            }
            buf = buf + r.value
        }
        body = "hello from RGC server"
        resp = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: " +
               to_string(body.len()) + "\r\n\r\n" + body
        s.send(resp)
        s.close()
        served = served + 1
    }
    listener.close()
    return 0
}

def main() -> int {
    argv = args()
    if argv.len() > 2 and argv.get(1) == "--serve" {
        return serve(argv.get(2))
    }

    port_file = "/tmp/rgc_http_demo.port"
    io.remove_file(port_file)
    cmd = "'" + argv.get(0) + "' --serve '" + port_file + "' >/dev/null 2>&1 &"
    os.system(cmd)

    # 等待服务器把端口写入文件
    port = 0
    tries = 0
    while tries < 200 {
        if io.exists(port_file) {
            fr = io.read_file(port_file)
            if fr.is_ok() {
                port = fr.unwrap().to_int().unwrap()
                break
            }
        }
        time.sleep(0.01)
        tries = tries + 1
    }
    if port == 0 {
        print("server did not start")
        return 1
    }
    print("server started on", port)

    base = "http://127.0.0.1:" + to_string(port)

    # GET：检查状态、头与正文
    r = http.get(base + "/hello")
    if r.ok {
        res = r.response
        print("status:", res.status)
        print("content-type:", http.header(@res, "Content-Type"))
        print("body:", res.body)
    } else {
        print("request failed:", r.error)
    }

    # POST
    p = http.post(base + "/echo", "hello")
    print("post:", p.ok, p.response.status)

    # 查询串编码
    pairs = List[str]()
    pairs.push("q=a b")
    pairs.push("x=1")
    print("encoded:", http.encode_query(pairs))

    return 0
}

运行输出：

server started on 38819
status: 200
content-type: text/plain
body: hello from RGC server
post: True 200
encoded: q=a%20b&x=1

这个例子把本篇的知识串起来了：

• net.listen_on + TcpListener.accept 实现服务器；
• TcpStream.recv / send 读写原始字节；
• io.write_file / io.read_file / io.exists 在父子进程间传递端口；
• os.system 启动后台进程；
• http.get / http.post / http.header / http.encode_query 完成客户端请求；
• 每个可能失败的点都用 ok 判断并优雅退出，没有 panic。

7. 注意事项

• 并发模型：同步网络调用是阻塞的。需要并发时用 async def + asyncio（一任务一线程），或用 event 事件循环 + 非阻塞原语在一个线程内管理大量 fd，也可自己派生子进程。
• 二进制数据用 Bytes：str 的 len() 基于 strlen，遇 NUL 会截断；send_bytes / recv_bytes / body_bytes / data_bytes 才是二进制安全的。
• 字符串生命周期：接收缓冲与返回字符串为堆分配，当前无 GC，进程退出时由系统回收。
• 错误即值：所有网络 API 通过 ok/error 返回失败原因，不要假设某个操作一定成功。
• 外部依赖：网络层只依赖 libc；TLS 与 gzip 均为纯 RGC 实现，不需要 OpenSSL/zlib。

8. 下一步

• 标准库规范：仓库 docs/compiler/STDLIB.md（网络章节含全部 API 细节与行为约定）
• RGC 入门教程：从 Hello World 到成绩管理器
• rack 入门：用 RGC 包管理器构建项目
• pyrgc 入门：把 RGC 函数变成 Python 原生扩展

RGC 以 Apache 2.0 协议开源。欢迎在 GitHub 提 Issue、贡献代码。