Skip to content

Commit a628936

Browse files
authored
refactor(core): implement FrameConn interface using quic-go (#642)
# Description Implement new two frame-related interfaces below using quic-go. ```go // Listener accepts FrameConns. type Listener interface { // Accept accepts FrameConns. Accept(context.Context) (FrameConn, error) // Close closes listener, // If listener be closed, all FrameConn accepted will be unavailable. Close() error } // FrameConn is a connection that transmits data in frame format. type FrameConn interface { // Context returns FrameConn.Context. // The Context can be used to manage the lifecycle of connection and // retrieve error using `context.Cause(conn.Context())` after calling `CloseWithError()`. Context() context.Context // WriteFrame writes a frame to connection. WriteFrame(frame.Frame) error // ReadFrame returns a channel from which frames can be received. ReadFrame() (frame.Frame, error) // RemoteAddr returns the remote address of connection. RemoteAddr() net.Addr // LocalAddr returns the local address of connection. LocalAddr() net.Addr // CloseWithError closes the connection with an error message. // It will be unavailable if the connection is closed. the error message should be written to the conn.Context(). CloseWithError(string) error } ```
1 parent 94370ab commit a628936

12 files changed

Lines changed: 539 additions & 562 deletions

File tree

core/client.go

Lines changed: 31 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -5,15 +5,14 @@ import (
55
"context"
66
"errors"
77
"fmt"
8-
"io"
98
"reflect"
109
"runtime"
1110
"time"
1211

13-
"github.com/quic-go/quic-go"
1412
"github.com/yomorun/yomo/core/frame"
1513
"github.com/yomorun/yomo/pkg/frame-codec/y3codec"
1614
"github.com/yomorun/yomo/pkg/id"
15+
yquic "github.com/yomorun/yomo/pkg/listener/quic"
1716
oteltrace "go.opentelemetry.io/otel/trace"
1817
"golang.org/x/exp/slog"
1918
)
@@ -69,35 +68,10 @@ func NewClient(appName, zipperAddr string, clientType ClientType, opts ...Client
6968
}
7069
}
7170

72-
type connectResult struct {
73-
conn quic.Connection
74-
fs *FrameStream
75-
err error
76-
}
77-
78-
func newConnectResult(conn quic.Connection, fs *FrameStream, err error) *connectResult {
79-
return &connectResult{
80-
conn: conn,
81-
fs: fs,
82-
err: err,
83-
}
84-
}
85-
86-
func (c *Client) connect(ctx context.Context) *connectResult {
87-
conn, err := quic.DialAddr(ctx, c.zipperAddr, c.opts.tlsConfig, c.opts.quicConfig)
88-
if err != nil {
89-
return newConnectResult(conn, nil, err)
90-
}
91-
92-
stream, err := conn.OpenStream()
71+
func (c *Client) connect(ctx context.Context, addr string) (frame.Conn, error) {
72+
conn, err := yquic.DialAddr(ctx, addr, y3codec.Codec(), y3codec.PacketReadWriter(), c.opts.tlsConfig, c.opts.quicConfig)
9373
if err != nil {
94-
return newConnectResult(conn, nil, err)
95-
}
96-
97-
fs := NewFrameStream(stream, y3codec.Codec(), y3codec.PacketReadWriter())
98-
99-
if credential := c.opts.credential; credential != nil {
100-
c.Logger.Info("use credential", "credential_name", credential.Name())
74+
return conn, err
10175
}
10276

10377
hf := &frame.HandshakeFrame{
@@ -109,79 +83,76 @@ func (c *Client) connect(ctx context.Context) *connectResult {
10983
AuthPayload: c.opts.credential.Payload(),
11084
}
11185

112-
if err := fs.WriteFrame(hf); err != nil {
113-
return newConnectResult(conn, nil, err)
86+
if err := conn.WriteFrame(hf); err != nil {
87+
return conn, err
11488
}
11589

116-
received, err := fs.ReadFrame()
90+
received, err := conn.ReadFrame()
11791
if err != nil {
118-
return newConnectResult(conn, nil, err)
92+
return nil, err
11993
}
120-
12194
switch received.Type() {
12295
case frame.TypeRejectedFrame:
123-
se := ErrAuthenticateFailed{received.(*frame.RejectedFrame).Message}
124-
return newConnectResult(conn, fs, se)
96+
return conn, ErrAuthenticateFailed{received.(*frame.RejectedFrame).Message}
12597
case frame.TypeHandshakeAckFrame:
126-
return newConnectResult(conn, fs, nil)
98+
return conn, nil
12799
default:
128-
se := ErrAuthenticateFailed{
100+
return conn, ErrAuthenticateFailed{
129101
fmt.Sprintf("authentication failed: read unexcepted frame, frame read: %s", received.Type().String()),
130102
}
131-
return newConnectResult(conn, fs, se)
132103
}
133104
}
134105

135-
func (c *Client) runBackground(ctx context.Context, conn quic.Connection, fs *FrameStream) {
106+
func (c *Client) runBackground(ctx context.Context, conn frame.Conn) {
136107
reconnection := make(chan struct{})
137108

138-
go c.handleReadFrames(fs, reconnection)
109+
go c.handleReadFrames(conn, reconnection)
139110

111+
var err error
140112
for {
141113
select {
142114
case <-c.ctx.Done():
143-
fs.Close()
115+
conn.CloseWithError("yomo: client closed")
144116
return
145117
case <-ctx.Done():
146-
fs.Close()
118+
conn.CloseWithError("yomo: parent context canceled")
147119
return
148120
case f := <-c.writeFrameChan:
149-
if err := fs.WriteFrame(f); err != nil {
121+
if err := conn.WriteFrame(f); err != nil {
150122
c.handleFrameError(err, reconnection)
151123
}
152124
case <-reconnection:
153125
reconnect:
154-
cr := c.connect(ctx)
155-
if err := cr.err; err != nil {
126+
conn, err = c.connect(ctx, c.zipperAddr)
127+
if err != nil {
156128
if errors.As(err, new(ErrAuthenticateFailed)) {
157129
return
158130
}
159-
c.Logger.Error("reconnect to zipper error", "err", cr.err)
131+
c.Logger.Error("reconnect to zipper error", "err", err)
160132
time.Sleep(time.Second)
161133
goto reconnect
162134
}
163-
fs = cr.fs
164-
go c.handleReadFrames(fs, reconnection)
135+
go c.handleReadFrames(conn, reconnection)
165136
}
166137
}
167138
}
168139

169140
// Connect connect client to server.
170141
func (c *Client) Connect(ctx context.Context) error {
171142
connect:
172-
result := c.connect(ctx)
173-
if result.err != nil {
143+
fconn, err := c.connect(ctx, c.zipperAddr)
144+
if err != nil {
174145
if c.opts.connectUntilSucceed {
175-
c.Logger.Error("failed to connect to zipper, trying to reconnect", "err", result.err)
146+
c.Logger.Error("failed to connect to zipper, trying to reconnect", "err", err)
176147
time.Sleep(time.Second)
177148
goto connect
178149
}
179-
c.Logger.Error("can not connect to zipper", "err", result.err)
180-
return result.err
150+
c.Logger.Error("can not connect to zipper", "err", err)
151+
return err
181152
}
182153
c.Logger.Info("connected to zipper")
183154

184-
go c.runBackground(ctx, result.conn, result.fs)
155+
go c.runBackground(ctx, fconn)
185156

186157
return nil
187158
}
@@ -238,8 +209,8 @@ func (c *Client) handleFrameError(err error, reconnection chan<- struct{}) {
238209
c.errorfn(err)
239210

240211
// exit client program if stream has be closed.
241-
if err == io.EOF {
242-
c.ctxCancel(fmt.Errorf("%s: remote shutdown", c.clientType.String()))
212+
if se := new(yquic.ErrConnClosed); errors.As(err, &se) {
213+
c.ctxCancel(fmt.Errorf("%s: shutdown with error=%s", c.clientType.String(), se.Error()))
243214
return
244215
}
245216

@@ -256,9 +227,9 @@ func (c *Client) Wait() {
256227
<-c.ctx.Done()
257228
}
258229

259-
func (c *Client) handleReadFrames(fs *FrameStream, reconnection chan struct{}) {
230+
func (c *Client) handleReadFrames(fconn frame.Conn, reconnection chan struct{}) {
260231
for {
261-
f, err := fs.ReadFrame()
232+
f, err := fconn.ReadFrame()
262233
if err != nil {
263234
c.handleFrameError(err, reconnection)
264235
return

core/connection.go

Lines changed: 6 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,6 @@
11
package core
22

33
import (
4-
"context"
5-
6-
"github.com/quic-go/quic-go"
74
"github.com/yomorun/yomo/core/frame"
85
"github.com/yomorun/yomo/core/metadata"
96
"golang.org/x/exp/slog"
@@ -31,42 +28,28 @@ type Connection struct {
3128
clientType ClientType
3229
metadata metadata.M
3330
observeDataTags []uint32
34-
conn quic.Connection
35-
fs *FrameStream
31+
fconn frame.Conn
3632
Logger *slog.Logger
3733
}
3834

3935
func newConnection(
4036
name string, id string, clientType ClientType, md metadata.M, tags []uint32,
41-
conn quic.Connection, fs *FrameStream, logger *slog.Logger) *Connection {
37+
fconn frame.Conn, logger *slog.Logger,
38+
) *Connection {
4239

4340
logger = logger.With("conn_id", id, "conn_name", name)
44-
if conn != nil {
45-
logger.Info("new client connected", "remote_addr", conn.RemoteAddr().String(), "client_type", clientType.String())
46-
}
4741

4842
return &Connection{
4943
name: name,
5044
id: id,
5145
clientType: clientType,
5246
metadata: md,
5347
observeDataTags: tags,
54-
conn: conn,
55-
fs: fs,
48+
fconn: fconn,
5649
Logger: logger,
5750
}
5851
}
5952

60-
// Close closes the connection.
61-
func (c *Connection) Close() error {
62-
return c.fs.Close()
63-
}
64-
65-
// Context returns the context of the connection.
66-
func (c *Connection) Context() context.Context {
67-
return c.fs.Context()
68-
}
69-
7053
// ID returns the connection ID.
7154
func (c *Connection) ID() string {
7255
return c.id
@@ -87,26 +70,10 @@ func (c *Connection) ObserveDataTags() []uint32 {
8770
return c.observeDataTags
8871
}
8972

90-
// ReadFrame reads a frame from the connection.
91-
func (c *Connection) ReadFrame() (frame.Frame, error) {
92-
return c.fs.ReadFrame()
93-
}
94-
95-
// ClientType returns the client type of the connection.
9673
func (c *Connection) ClientType() ClientType {
9774
return c.clientType
9875
}
9976

100-
// WriteFrame writes a frame to the connection.
101-
func (c *Connection) WriteFrame(f frame.Frame) error {
102-
return c.fs.WriteFrame(f)
77+
func (c *Connection) FrameConn() frame.Conn {
78+
return c.fconn
10379
}
104-
105-
// CloseWithError closes the connection with error.
106-
func (c *Connection) CloseWithError(errString string) error {
107-
return c.conn.CloseWithError(YomoCloseErrorCode, errString)
108-
}
109-
110-
// YomoCloseErrorCode is the error code for close quic Connection for yomo.
111-
// If the Connection implemented by quic is closed, the quic ApplicationErrorCode is always 0x13.
112-
const YomoCloseErrorCode = quic.ApplicationErrorCode(0x13)

0 commit comments

Comments
 (0)