kitex
kitex copied to clipboard
cloudwego
使用 server.WithMuxTransport() 时报错
NewService() 启用 server.WithMuxTransport() 后第一次调用时报错
To Reproduce
Steps to reproduce the behavior:
服务端代码:
func main() {
r, err := etcd.NewEtcdRegistry([]string{
`127.0.0.1:2379`,
}, etcd.WithAuthOpt(`user`, `passwd`),
etcd.WithDialTimeoutOpt(2*time.Second))
if err != nil {
log.Panic(err)
}
svr := user_provider.NewServer(new(UserProviderImpl),
server.WithRegistry(r), server.WithRegistryInfo(®istry.Info{
ServiceName: `DemoProvider`,
WarmUp: 2 * time.Second,
}), server.WithMuxTransport())
err = svr.Run()
if err != nil {
log.Println(err.Error())
}
客户端代码:
func main() {
r, e := etcd.NewEtcdResolver([]string{
`127.0.0.1:2379`,
}, etcd.WithAuthOpt(`user`, `passwd`), etcd.WithDialTimeoutOpt(2*time.Second))
if e != nil {
log.Panic(e)
}
client := userprovider.MustNewClient(`DemoProvider`, client2.WithResolver(r), client2.WithMuxConnection(10))
for i := 0; i < 60000; i++ {
u, e := client.Get(context.Background(), int64(i))
if e != nil {
log.Printf("call failed, return's error: %v", e)
}
log.Printf("%#v", *u)
}
}
Expected behavior
代码正常运行
Screenshots
2023/09/13 20:15:16.592946 server.go:83: [Info] KITEX: server listen at addr=[::]:8888
2023/09/13 20:15:17.639616 etcd_registry.go:171: [Info] start keepalive lease 694d8a8e53e07c6e for etcd registry
panic: interface conversion: *gonet.bufioConn is not netpoll.Connection: missing method AddCloseCallback
goroutine 58 [running]:
github.com/cloudwego/kitex/pkg/remote/trans/netpollmux.(*svrTransHandler).OnActive(0xc00040ec80, {0x276f230d798?, 0xc00040ec80?}, {0x11808a8?, 0xc000070500})
C:/Users/my/go/pkg/mod/github.com/cloudwego/[email protected]/pkg/remote/trans/netpollmux/server_handler.go:292 +0x49
github.com/cloudwego/kitex/pkg/remote.(*TransPipeline).OnActive(0xc0002d2bc0, {0x117c438?, 0x17c92c0?}, {0x11808a8, 0xc000070500})
C:/Users/my/go/pkg/mod/github.com/cloudwego/[email protected]/pkg/remote/trans_pipeline.go:106 +0xda
github.com/cloudwego/kitex/pkg/remote/trans/gonet.(*transServer).BootstrapServer.func1()
C:/Users/my/go/pkg/mod/github.com/cloudwego/[email protected]/pkg/remote/trans/gonet/trans_server.go:94 +0xe5
created by github.com/cloudwego/kitex/pkg/remote/trans/gonet.(*transServer).BootstrapServer in goroutine 52
C:/Users/my/go/pkg/mod/github.com/cloudwego/[email protected]/pkg/remote/trans/gonet/trans_server.go:85 +0x88
Kitex version:
Kitex v0.7.1
Environment:
set GO111MODULE=
set GOARCH=amd64
set GOBIN=
set GOCACHE=C:\Users\my\AppData\Local\go-build
set GOENV=C:\Users\my\AppData\Roaming\go\env
set GOEXE=.exe
set GOEXPERIMENT=
set GOFLAGS=
set GOHOSTARCH=amd64
set GOHOSTOS=windows
set GOINSECURE=
set GOMODCACHE=C:\Users\my\go\pkg\mod
set GONOPROXY=
set GONOSUMDB=
set GOOS=windows
set GOPATH=C:\Users\my\go
set GOPRIVATE=
set GOPROXY=https://goproxy.cn,direct
set GOROOT=C:\Program Files\Go
set GOSUMDB=sum.golang.org
set GOTMPDIR=
set GOTOOLCHAIN=auto
set GOTOOLDIR=C:\Program Files\Go\pkg\tool\windows_amd64
set GOVCS=
set GOVERSION=go1.21.1
set GCCGO=gccgo
set GOAMD64=v1
set AR=ar
set CC=gcc
set CXX=g++
set CGO_ENABLED=1
set GOWORK=
set CGO_CFLAGS=-O2 -g
set CGO_CPPFLAGS=
set CGO_CXXFLAGS=-O2 -g
set CGO_FFLAGS=-O2 -g
set CGO_LDFLAGS=-O2 -g
set PKG_CONFIG=pkg-config
set GOGCCFLAGS=-m64 -mthreads -Wl,--no-gc-sections -fmessage-length=0 -ffile-prefix-map=C:\Users\my\AppData\Local\Temp\go-build803072253=/tmp/go-build -gno-record-gcc-switches
Additional context
Add any other context about the problem here.
这是代码bug
gonet.bufioConn （定义在github.com/cloudwego/kitex/pkg/remote/trans/gonet/trans_server.go）没有实现 AddCloseCallback
gonet.bufioConn 的定义:
// bufioConn implements the net.Conn interface.
type bufioConn struct {
conn net.Conn
r netpoll.Reader
}
func newBufioConn(c net.Conn) *bufioConn {
return &bufioConn{
conn: c,
r: netpoll.NewReader(c),
}
}
func (bc *bufioConn) RawConn() net.Conn {
return bc.conn
}
func (bc *bufioConn) Read(b []byte) (int, error) {
buf, err := bc.r.Next(len(b))
if err != nil {
return 0, err
}
copy(b, buf)
return len(b), nil
}
func (bc *bufioConn) Write(b []byte) (int, error) {
return bc.conn.Write(b)
}
func (bc *bufioConn) Close() error {
bc.r.Release()
return bc.conn.Close()
}
func (bc *bufioConn) LocalAddr() net.Addr {
return bc.conn.LocalAddr()
}
func (bc *bufioConn) RemoteAddr() net.Addr {
return bc.conn.RemoteAddr()
}
func (bc *bufioConn) SetDeadline(t time.Time) error {
return bc.conn.SetDeadline(t)
}
func (bc *bufioConn) SetReadDeadline(t time.Time) error {
return bc.conn.SetReadDeadline(t)
}
func (bc *bufioConn) SetWriteDeadline(t time.Time) error {
return bc.conn.SetWriteDeadline(t)
}
func (bc *bufioConn) Reader() netpoll.Reader {
return bc.r
}
而 netpoll.Connection 的声明为:
// CloseCallback will be called when the connection is closed.
// Return: error is unused which will be ignored directly.
type CloseCallback func(connection Connection) error
// Connection supports reading and writing simultaneously,
// but does not support simultaneous reading or writing by multiple goroutines.
// It maintains its own input/output buffer, and provides nocopy API for reading and writing.
type Connection interface {
// Connection extends net.Conn, just for interface compatibility.
// It's not recommended to use net.Conn API except for io.Closer.
net.Conn
// The recommended API for nocopy reading and writing.
// Reader will return nocopy buffer data, or error after timeout which set by SetReadTimeout.
Reader() Reader
// Writer will write data to the connection by NIO mode,
// so it will return an error only when the connection isn't Active.
Writer() Writer
// IsActive checks whether the connection is active or not.
IsActive() bool
// SetReadTimeout sets the timeout for future Read calls wait.
// A zero value for timeout means Reader will not timeout.
SetReadTimeout(timeout time.Duration) error
// SetWriteTimeout sets the timeout for future Write calls wait.
// A zero value for timeout means Writer will not timeout.
SetWriteTimeout(timeout time.Duration) error
// SetIdleTimeout sets the idle timeout of connections.
// Idle connections that exceed the set timeout are no longer guaranteed to be active,
// but can be checked by calling IsActive.
SetIdleTimeout(timeout time.Duration) error
// SetOnRequest can set or replace the OnRequest method for a connection, but can't be set to nil.
// Although SetOnRequest avoids data race, it should still be used before transmitting data.
// Replacing OnRequest while processing data may cause unexpected behavior and results.
// Generally, the server side should uniformly set the OnRequest method for each connection via NewEventLoop,
// which is set when the connection is initialized.
// On the client side, if necessary, make sure that OnRequest is set before sending data.
SetOnRequest(on OnRequest) error
// AddCloseCallback can add hangup callback for a connection, which will be called when connection closing.
// This is very useful for cleaning up idle connections. For instance, you can use callbacks to clean up
// the local resources, which bound to the idle connection, when hangup by the peer. No need another goroutine
// to polling check connection status.
AddCloseCallback(callback CloseCallback) error
}
windows 不支持WithMuxTransport()
@zhangsilly 是的，windows 下目前是 fallback 到 go net，对 mux 不支持
