diff --git a/client.go b/client.go index 712dc40..39b92c0 100644 --- a/client.go +++ b/client.go @@ -25,9 +25,9 @@ type Client struct { opts *clientOptions // ctx and ctxCancel manage the lifecycle of client. - ctx context.Context - ctxCancel context.CancelFunc - + ctx context.Context + ctxCancel context.CancelFunc + logger log.Logger writeFrameChan chan frame.Frame shutdownChan chan error } @@ -47,24 +47,29 @@ func NewClient(appName string, connType ClientType, opts ...ClientOption) *Clien ctx, ctxCancel := context.WithCancel(context.Background()) - 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 := new(Client) + if cli.logger == nil { + cli.logger = log.Default() } + cli.name = appName + cli.clientId = clientId + cli.streamType = connType + cli.opts = option + cli.errorfn = func(err error) { + cli.logger.Errorf("client err, %s", 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 { - log.Errorf("connect error, %s", err) + c.logger.Errorf("connect error, %s", err) return err } @@ -91,7 +96,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 { - log.Errorf("client reconnect error, %s", err) + c.logger.Errorf("client reconnect error, %s", err) time.Sleep(time.Second) goto RECONNECT } @@ -110,7 +115,7 @@ func (c *Client) cleanStream(controlStream ClientControlStream, err error) { errString := "" if err != nil { errString = err.Error() - log.Errorf("client cancel with error, %s", err) + c.logger.Errorf("client cancel with error, %s", err) } // controlStream is nil represents that client is not connected. @@ -250,18 +255,18 @@ func (c *Client) handleFrame(f frame.Frame) { switch ff := f.(type) { case *frame.DataFrame: if c.processor == nil { - log.Warnf("client processor has not been set") + c.logger.Warnf("client processor has not been set") } else { c.processor(ff) } case *frame.BackFlowFrame: if c.receiver == nil { - log.Warnf("client receiver has not been set") + c.logger.Warnf("client receiver has not been set") } else { c.receiver(ff) } default: - log.Warnf("client data stream receive unexcepted frame, frame_type: %v", f) + c.logger.Warnf("client data stream receive unexcepted frame, frame_type: %v", f) } } @@ -291,13 +296,13 @@ 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 - log.Debugf("SetDataFrameObserver") + c.logger.Debugf("SetDataFrameObserver") } // SetBackFlowFrameObserver sets the backflow frame handler. func (c *Client) SetBackFlowFrameObserver(fn func(*frame.BackFlowFrame)) { c.receiver = fn - log.Debugf("SetBackFlowFrameObserver") + c.logger.Debugf("SetBackFlowFrameObserver") } // SetObserveDataTags set the data tag list that will be observed. @@ -315,3 +320,8 @@ 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 +}