Component
/
network
3
0
Fork 0
Code Issues Pull Requests Projects Releases 9 Wiki Activity

Compare commits

base: Component:master
Branches Tags
Component:master
Component:v1.11
Component:v1.10
Component:v1.9
Component:v1.8
Component:v1.7
Component:v1.6
Component:v1.5
Component:v1.4
Component:v.1.3
..
compare: Component:v1.6
Branches Tags
Component:master
Component:v1.11
Component:v1.10
Component:v1.9
Component:v1.8
Component:v1.7
Component:v1.6
Component:v1.5
Component:v1.4
Component:v.1.3

No commits in common. "master" and "v1.6" have entirely different histories.
master ... v1.6

9 changed files with 82 additions and 101 deletions
Show Stats Expand all files Collapse all files

56
client.go
View File

@ -25,9 +25,9 @@ type Client struct {
opts *clientOptions
// ctx and ctxCancel manage the lifecycle of client.
ctx context.Context
ctxCancel context.CancelFunc
logger log.Logger
ctx context.Context
ctxCancel context.CancelFunc
writeFrameChan chan frame.Frame
shutdownChan chan error
}
@ -42,34 +42,29 @@ func NewClient(appName string, connType ClientType, opts ...ClientOption) *Clien
clientId := id.New()
if option.credential != nil {
log.Infof("%s use credential, credential_name: %s;", ClientLogPrefix, option.credential.Name())
log.Infof("use credential, credential_name: %s;", option.credential.Name())
}
ctx, ctxCancel := context.WithCancel(context.Background())
cli := new(Client)
if cli.logger == nil {
cli.logger = log.Default()
return &Client{
name: appName,
clientId: clientId,
streamType: connType,
opts: option,
errorfn: func(err error) { log.Errorf("client err, %s", err) },
writeFrameChan: make(chan frame.Frame),
shutdownChan: make(chan error, 1),
ctx: ctx,
ctxCancel: ctxCancel,
}
cli.name = appName
cli.clientId = clientId
cli.streamType = connType
cli.opts = option
cli.errorfn = func(err error) {
cli.logger.Errorf("%s client err, %s", ClientLogPrefix, err)
}
cli.writeFrameChan = make(chan frame.Frame)
cli.shutdownChan = make(chan error, 1)
cli.ctx = ctx
cli.ctxCancel = ctxCancel
return cli
}
// Connect connects to HPDS-MessageQueue.
func (c *Client) Connect(ctx context.Context, addr string) error {
controlStream, dataStream, err := c.openStream(ctx, addr)
if err != nil {
c.logger.Errorf("%s connect error, %s", ClientLogPrefix, err)
log.Errorf("connect error, %s", err)
return err
}
@ -96,7 +91,7 @@ func (c *Client) runBackground(ctx context.Context, addr string, controlStream C
var err error
controlStream, dataStream, err = c.openStream(ctx, addr)
if err != nil {
c.logger.Errorf("%s client reconnect error, %s", ClientLogPrefix, err)
log.Errorf("client reconnect error, %s", err)
time.Sleep(time.Second)
goto RECONNECT
}
@ -115,7 +110,7 @@ func (c *Client) cleanStream(controlStream ClientControlStream, err error) {
errString := ""
if err != nil {
errString = err.Error()
c.logger.Errorf("%s client cancel with error, %s", ClientLogPrefix, err)
log.Errorf("client cancel with error, %s", err)
}
// controlStream is nil represents that client is not connected.
@ -255,18 +250,18 @@ func (c *Client) handleFrame(f frame.Frame) {
switch ff := f.(type) {
case *frame.DataFrame:
if c.processor == nil {
c.logger.Warnf("%s client processor has not been set", ClientLogPrefix)
log.Warnf("client processor has not been set")
} else {
c.processor(ff)
}
case *frame.BackFlowFrame:
if c.receiver == nil {
c.logger.Warnf("%s client receiver has not been set", ClientLogPrefix)
log.Warnf("client receiver has not been set")
} else {
c.receiver(ff)
}
default:
c.logger.Warnf("%s client data stream receive unexcepted frame, frame_type: %v", ClientLogPrefix, f)
log.Warnf("client data stream receive unexcepted frame, frame_type: %v", f)
}
}
@ -296,17 +291,17 @@ func (c *Client) receivingStreamClose(controlStream ControlStream, dataStream Da
// SetDataFrameObserver sets the data frame handler.
func (c *Client) SetDataFrameObserver(fn func(*frame.DataFrame)) {
c.processor = fn
c.logger.Debugf("%s SetDataFrameObserver", ClientLogPrefix)
log.Debugf("SetDataFrameObserver")
}
// SetBackFlowFrameObserver sets the backflow frame handler.
func (c *Client) SetBackFlowFrameObserver(fn func(*frame.BackFlowFrame)) {
c.receiver = fn
c.logger.Debugf("%s SetBackFlowFrameObserver", ClientLogPrefix)
log.Debugf("SetBackFlowFrameObserver")
}
// SetObserveDataTags set the data tag list that will be observed.
// Deprecated: use network.WithObserveDataTags instead
// Deprecated: use yomo.WithObserveDataTags instead
func (c *Client) SetObserveDataTags(tag ...frame.Tag) {
c.opts.observeDataTags = append(c.opts.observeDataTags, tag...)
}
@ -320,8 +315,3 @@ func (c *Client) SetErrorHandler(fn func(err error)) {
func (c *Client) ClientId() string {
return c.clientId
}
// Logger get client's logger instance, you can customize this using `hpds.WithLogger`
func (c *Client) Logger() log.Logger {
return c.logger
}

4
connector.go
View File

@ -41,7 +41,7 @@ func (c *Connector) Add(streamId string, stream DataStream) error {
c.streams.Store(streamId, stream)
log.Debugf("%s Connector add stream, stream_id: %s", ParseFrameLogPrefix, streamId)
log.Debugf("Connector add stream, stream_id: %s", streamId)
return nil
}
@ -55,7 +55,7 @@ func (c *Connector) Remove(streamId string) error {
}
c.streams.Delete(streamId)
log.Debugf("%s Connector remove stream, stream_id: %s", ParseFrameLogPrefix, streamId)
log.Debugf("Connector remove stream, stream_id: %s", streamId)
return nil
}

7
constant.go
View File

@ -17,13 +17,6 @@ const (
ConnStateClosed ConnState = "Closed"
)
// Prefix is the prefix for logger.
const (
ClientLogPrefix = "\033[36m[network:client]\033[0m "
ServerLogPrefix = "\033[32m[network:server]\033[0m "
ParseFrameLogPrefix = "\033[36m[network:stream_parser]\033[0m "
)
func init() {
rand.Seed(time.Now().Unix())
}

6
context.go
View File

@ -84,7 +84,7 @@ func newContext(dataStream DataStream) (c *Context) {
c = v.(*Context)
}
log.Infof("%s stream_id: %s; stream_name: %s; stream_type: %s;", ClientLogPrefix, dataStream.ID(),
log.Infof("stream_id: %s; stream_name: %s; stream_type: %s;", dataStream.ID(),
dataStream.Name(), dataStream.StreamType().String(),
)
@ -105,13 +105,13 @@ func (c *Context) WithFrame(f frame.Frame) {
//
// TODO: ycode is not be transmitted.
func (c *Context) CloseWithError(hCode hpds_err.ErrorCode, errString string) {
log.Warnf("%s Stream Close With error, err_code: %s; error: %s", ClientLogPrefix, hCode.String(), errString)
log.Warnf("Stream Close With error", "err_code", hCode.String(), "error", errString)
err := c.DataStream.CloseWithError(errString)
if err == nil {
return
}
log.Errorf("%s Close DataStream error %v", ClientLogPrefix, err)
log.Errorf("Close DataStream error", err)
}
// Clean cleans the Context,

2
handler_type.go
View File

@ -3,7 +3,7 @@ package network
import "git.hpds.cc/Component/network/frame"
// AsyncHandler is the request-response mode (async)
type AsyncHandler func([]byte) (frame.Tag, []byte)
type AsyncHandler func([]byte) (byte, []byte)
// PipeHandler is the bidirectional stream mode (blocking).
type PipeHandler func(in <-chan []byte, out chan<- *frame.PayloadFrame)

2
log/logger.go
View File

@ -120,7 +120,7 @@ func logFormat() string {
func logLevel() Level {
envLevel := strings.ToLower(os.Getenv("HPDS_LOG_LEVEL"))
level := DebugLevel
level := ErrorLevel
switch envLevel {
case "debug":
return DebugLevel

28
router/default.go
View File

@ -2,9 +2,11 @@
package router
import (
"fmt"
"sync"
"git.hpds.cc/Component/network/frame"
herr "git.hpds.cc/Component/network/hpds_err"
"git.hpds.cc/Component/network/metadata"
)
@ -59,23 +61,19 @@ func (r *defaultRoute) Add(connId string, name string, observeDataTags []frame.T
}
}
if !ok {
//return fmt.Errorf("SFN[%s] does not exist in config functions", name)
//item := config.App{
// Name: name,
//}
r.functions = append(r.functions, name)
return fmt.Errorf("SFN[%s] does not exist in config functions", name)
}
//LOOP:
//for _, conn := range r.data {
// for connId, n := range conn {
// if n == name {
// err = herr.NewDuplicateNameError(connId, fmt.Errorf("SFN[%s] is already linked to another connection", name))
// delete(conn, connId)
// break LOOP
// }
// }
//}
LOOP:
for _, conn := range r.data {
for connId, n := range conn {
if n == name {
err = herr.NewDuplicateNameError(connId, fmt.Errorf("SFN[%s] is already linked to another connection", name))
delete(conn, connId)
break LOOP
}
}
}
for _, tag := range observeDataTags {
conn := r.data[tag]

76
server.go
View File

@ -42,7 +42,6 @@ type Server struct {
afterHandlers []FrameHandler
connectionCloseHandlers []ConnectionHandler
listener Listener
logger log.Logger
}
// NewServer create a Server instance.
@ -86,21 +85,21 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error {
if err := s.validateRouter(); err != nil {
return err
}
s.connector = NewConnector(ctx)
// listen the address
listener, err := newListener(conn, s.opts.tlsConfig, s.opts.quicConfig)
if err != nil {
log.Errorf("%s listener.Listen: err=%v", ServerLogPrefix, err)
log.Errorf("listener.Listen: err=%v", err)
return err
}
s.listener = listener
log.Printf("%s [%s][%d] Listening on: %s, QUIC: %v, AUTH: %s", ServerLogPrefix, s.name, os.Getpid(), listener.Addr(), listener.Versions(), s.authNames())
log.Printf("[%s][%d] Listening on: %s, QUIC: %v, AUTH: %s", s.name, os.Getpid(), listener.Addr(), listener.Versions(), s.authNames())
for {
conn, err := s.listener.Accept(ctx)
if err != nil {
log.Errorf("%s listener accept connections error %v", ServerLogPrefix, err)
log.Errorf("listener accept connections error", err)
return err
}
err = s.opts.alpnHandler(conn.ConnectionState().TLS.NegotiatedProtocol)
@ -120,10 +119,10 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error {
// It response to client a AuthenticationAckFrame.
err = controlStream.VerifyAuthentication(s.handleAuthenticationFrame)
if err != nil {
log.Warnf("%s Authentication Failed, error: %s", ServerLogPrefix, err)
log.Warnf("Authentication Failed", "error", err)
continue
}
log.Debugf("%s Authentication Success", ServerLogPrefix)
log.Debugf("Authentication Success")
go func(qConn quic.Connection) {
streamGroup := NewStreamGroup(ctx, controlStream, s.connector)
@ -135,7 +134,7 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error {
case <-ctx.Done():
return
case err := <-s.runWithStreamGroup(streamGroup):
log.Errorf("%s Client Close, %v", ServerLogPrefix, err)
log.Errorf("Client Close, %v", err)
}
}(conn)
}
@ -184,8 +183,8 @@ func (s *Server) handleRoute(c *Context) error {
if e, ok := err.(hpds_err.DuplicateNameError); ok {
existsConnId := e.ConnId()
log.Debugf("%s StreamFunction Duplicate Name, error: %s; sfn_name: %s, old_stream_id: %s; current_stream_id: %s",
ServerLogPrefix, e.Error(), c.DataStream.Name(), existsConnId, c.StreamId())
log.Debugf("StreamFunction Duplicate Name, error: %s; sfn_name: %s, old_stream_id: %s; current_stream_id: %s",
e.Error(), c.DataStream.Name(), existsConnId, c.StreamId())
stream, ok, err := s.connector.Get(existsConnId)
if err != nil {
@ -215,11 +214,12 @@ func (s *Server) handleStreamContext(c *Context) {
// start frame handlers
for _, handler := range s.startHandlers {
if err := handler(c); err != nil {
log.Errorf("%s startHandlers error: %v", ServerLogPrefix, err)
log.Errorf("startHandlers error: %v", err)
c.CloseWithError(hpds_err.ErrorCodeStartHandler, err.Error())
return
}
}
// check update for stream
for {
f, err := c.DataStream.ReadFrame()
@ -228,26 +228,26 @@ func (s *Server) handleStreamContext(c *Context) {
if e, ok := err.(*quic.ApplicationError); ok {
if hpds_err.Is(e.ErrorCode, hpds_err.ErrorCodeClientAbort) {
// client abort
log.Infof("%s client close the connection", ServerLogPrefix)
log.Infof("client close the connection")
break
}
he := hpds_err.New(hpds_err.Parse(e.ErrorCode), err)
log.Errorf("%s read frame error: %v", ServerLogPrefix, he)
log.Errorf("read frame error: %v", he)
} else if err == io.EOF {
log.Infof("%s connection EOF", ServerLogPrefix)
log.Infof("connection EOF")
break
}
if errors.Is(err, net.ErrClosed) {
// if client close the connection, net.ErrClosed will be raise
// by quic-go IdleTimeoutError after connection's KeepAlive config.
log.Warnf("%s connection error, error: %v", ServerLogPrefix, net.ErrClosed)
log.Warnf("connection error, error: %v", net.ErrClosed)
c.CloseWithError(hpds_err.ErrorCodeClosed, "net.ErrClosed")
break
}
// any error occurred, we should close the stream
// after this, conn.AcceptStream() will raise the error
c.CloseWithError(hpds_err.ErrorCodeUnknown, err.Error())
log.Warnf("%s connection close", ServerLogPrefix)
log.Warnf("connection close")
break
}
@ -257,21 +257,21 @@ func (s *Server) handleStreamContext(c *Context) {
// before frame handlers
for _, handler := range s.beforeHandlers {
if err := handler(c); err != nil {
log.Errorf("%s beforeFrameHandler error: %v", ServerLogPrefix, err)
log.Errorf("beforeFrameHandler error: %v", err)
c.CloseWithError(hpds_err.ErrorCodeBeforeHandler, err.Error())
return
}
}
// main handler
if err := s.mainFrameHandler(c); err != nil {
log.Errorf("%s mainFrameHandler error: %v", ServerLogPrefix, err)
log.Errorf("mainFrameHandler error: %v", err)
c.CloseWithError(hpds_err.ErrorCodeMainHandler, err.Error())
return
}
// after frame handler
for _, handler := range s.afterHandlers {
if err := handler(c); err != nil {
log.Errorf("%s afterFrameHandler error: %v", ServerLogPrefix, err)
log.Errorf("afterFrameHandler error: %v", err)
c.CloseWithError(hpds_err.ErrorCodeAfterHandler, err.Error())
return
}
@ -293,7 +293,7 @@ func (s *Server) mainFrameHandler(c *Context) error {
_ = s.handleBackFlowFrame(c)
}
default:
log.Errorf("%s err=%v, frame=%v", ServerLogPrefix, err, frame.Shortly(c.Frame.Encode()))
log.Errorf("err=%v, frame=%v", err, frame.Shortly(c.Frame.Encode()))
}
return nil
}
@ -302,9 +302,9 @@ func (s *Server) handleAuthenticationFrame(f auth.Object) (bool, error) {
ok := auth.Authenticate(s.opts.auths, f)
if ok {
log.Debugf("%s Successful authentication", ServerLogPrefix)
log.Debugf("Successful authentication")
} else {
log.Warnf("%s Authentication failed, credential: %s", ServerLogPrefix, f.AuthName())
log.Warnf("Authentication failed", "credential", f.AuthName())
}
return ok, nil
@ -319,7 +319,7 @@ func (s *Server) handleDataFrame(c *Context) error {
return err
}
if !ok {
log.Warnf("%s handleDataFrame connector cannot find, from_conn_id: %s", ServerLogPrefix, fromId)
log.Warnf("handleDataFrame connector cannot find, from_conn_id: %s", fromId)
return fmt.Errorf("handleDataFrame connector cannot find %s", fromId)
}
@ -340,14 +340,14 @@ func (s *Server) handleDataFrame(c *Context) error {
// route
route := s.router.Route(m)
if route == nil {
log.Warnf("%s handleDataFrame route is nil", ServerLogPrefix)
log.Warnf("handleDataFrame route is nil")
return fmt.Errorf("handleDataFrame route is nil")
}
// get stream function connection ids from route
connIDs := route.GetForwardRoutes(f.GetDataTag())
log.Debugf("%s Data Routing Status, sfn_stream_ids: %v; connector: %v", ServerLogPrefix, connIDs, s.connector.GetSnapshot())
log.Debugf("Data Routing Status, sfn_stream_ids: %v; connector: %v", connIDs, s.connector.GetSnapshot())
for _, toId := range connIDs {
conn, ok, err := s.connector.Get(toId)
@ -355,17 +355,17 @@ func (s *Server) handleDataFrame(c *Context) error {
continue
}
if !ok {
log.Errorf("%s Can't find forward conn, error: conn is nil ;forward_conn_id: ", ServerLogPrefix, toId)
log.Errorf("Can't find forward conn, error: conn is nil ;forward_conn_id: ", toId)
continue
}
to := conn.Name()
log.Infof("%s handleDataFrame, from_conn_name: %s; from_conn_id: %s; to_conn_name: %s; to_conn_id: %s; data_frame: %s",
ServerLogPrefix, from.Name(), fromId, to, toId, f.String())
log.Infof("handleDataFrame, from_conn_name: %s; from_conn_id: %s; to_conn_name: %s; to_conn_id: %s; data_frame: %s",
from.Name(), fromId, to, toId, f.String())
// write data frame to stream
if err := conn.WriteFrame(f); err != nil {
log.Errorf("%s handleDataFrame conn.Write, %v", ServerLogPrefix, err)
log.Errorf("handleDataFrame conn.Write, %v", err)
}
}
@ -386,9 +386,9 @@ func (s *Server) handleBackFlowFrame(c *Context) error {
// logger.Printf("%s handleBackFlowFrame tag:%#v --> source:%s, result=%s", ServerLogPrefix, tag, sourceId, carriage)
for _, source := range sourceConnList {
if source != nil {
log.Debugf("%s handleBackFlowFrame tag:%#v; source_conn_id: %s, back_flow_frame: %s", ServerLogPrefix, tag, sourceId, f.String())
log.Debugf("handleBackFlowFrame tag:%#v; source_conn_id: %s, back_flow_frame: %s", tag, sourceId, f.String())
if err := source.WriteFrame(bf); err != nil {
log.Errorf("%s handleBackFlowFrame tag:%#v --> Protocol Gateway:%s, error=%v", ServerLogPrefix, tag, sourceId, err)
log.Errorf("handleBackFlowFrame tag:%#v --> Protocol Gateway:%s, error=%v", tag, sourceId, err)
return err
}
}
@ -415,7 +415,7 @@ func (s *Server) DownStreams() map[string]frame.Writer {
func (s *Server) ConfigRouter(router router.Router) {
s.mu.Lock()
s.router = router
log.Debugf("%s config router is %#v", ServerLogPrefix, router)
log.Debugf("config router is %#v", router)
s.mu.Unlock()
}
@ -424,7 +424,7 @@ func (s *Server) ConfigRouter(router router.Router) {
func (s *Server) ConfigMetadataBuilder(builder metadata.Builder) {
s.mu.Lock()
s.metadataBuilder = builder
log.Debugf("%s config metadataBuilder is %#v", ServerLogPrefix, builder)
log.Debugf("config metadataBuilder is %#v", builder)
s.mu.Unlock()
}
@ -432,7 +432,7 @@ func (s *Server) ConfigMetadataBuilder(builder metadata.Builder) {
func (s *Server) ConfigAlpnHandler(h func(string) error) {
s.mu.Lock()
s.opts.alpnHandler = h
log.Debugf("%s config alpnHandler", ServerLogPrefix)
log.Debugf("config alpnHandler")
s.mu.Unlock()
}
@ -448,11 +448,11 @@ func (s *Server) AddDownstreamServer(addr string, c *Client) {
func (s *Server) dispatchToDownStreams(c *Context) {
stream, ok, err := s.connector.Get(c.StreamId())
if err != nil {
log.Errorf("%s Connector Get Error, %v", ServerLogPrefix, err)
log.Errorf("Connector Get Error, %v", err)
return
}
if !ok {
log.Debugf("%s dispatchTo Down Streams failed", ServerLogPrefix)
log.Debugf("dispatchTo Down Streams failed")
return
}
@ -463,7 +463,7 @@ func (s *Server) dispatchToDownStreams(c *Context) {
f.GetMetaFrame().SetMetadata(stream.Metadata())
}
for addr, ds := range s.downStreams {
log.Infof("%s dispatching to, dispatch_addr: %s; tid: %s;", ServerLogPrefix, addr, "", f.TransactionId())
log.Infof("dispatching to, dispatch_addr: %s; tid: %s;", addr, "", f.TransactionId())
_ = ds.WriteFrame(f)
}
}
@ -522,7 +522,7 @@ func (s *Server) authNames() []string {
}
func (s *Server) doConnectionCloseHandlers(qConn quic.Connection) {
log.Debugf("%s QUIC Connection Closed", ServerLogPrefix)
log.Debugf("QUIC Connection Closed")
for _, h := range s.connectionCloseHandlers {
h(qConn)
}

2
server_options.go
View File

@ -32,7 +32,7 @@ func defaultServerOptions() *serverOptions {
auths: map[string]auth.Authentication{},
}
opts.alpnHandler = func(proto string) error {
log.Infof("%s client alpn proto %s, %s, %s, %s", ServerLogPrefix, "component", "server", "proto", proto)
log.Infof("client alpn proto", "component", "server", "proto", proto)
return nil
}
return opts