增加客户端日志输出
This commit is contained in:
		
							parent
							
								
									2ba9e9c0eb
								
							
						
					
					
						commit
						4a4fdbd4f6
					
				
							
								
								
									
										52
									
								
								client.go
								
								
								
								
							
							
						
						
									
										52
									
								
								client.go
								
								
								
								
							|  | @ -25,9 +25,9 @@ type Client struct { | ||||||
| 	opts       *clientOptions | 	opts       *clientOptions | ||||||
| 
 | 
 | ||||||
| 	// ctx and ctxCancel manage the lifecycle of client.
 | 	// ctx and ctxCancel manage the lifecycle of client.
 | ||||||
| 	ctx       context.Context | 	ctx            context.Context | ||||||
| 	ctxCancel context.CancelFunc | 	ctxCancel      context.CancelFunc | ||||||
| 
 | 	logger         log.Logger | ||||||
| 	writeFrameChan chan frame.Frame | 	writeFrameChan chan frame.Frame | ||||||
| 	shutdownChan   chan error | 	shutdownChan   chan error | ||||||
| } | } | ||||||
|  | @ -47,24 +47,29 @@ func NewClient(appName string, connType ClientType, opts ...ClientOption) *Clien | ||||||
| 
 | 
 | ||||||
| 	ctx, ctxCancel := context.WithCancel(context.Background()) | 	ctx, ctxCancel := context.WithCancel(context.Background()) | ||||||
| 
 | 
 | ||||||
| 	return &Client{ | 	cli := new(Client) | ||||||
| 		name:           appName, | 	if cli.logger == nil { | ||||||
| 		clientId:       clientId, | 		cli.logger = log.Default() | ||||||
| 		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("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.
 | // Connect connects to HPDS-MessageQueue.
 | ||||||
| func (c *Client) Connect(ctx context.Context, addr string) error { | func (c *Client) Connect(ctx context.Context, addr string) error { | ||||||
| 	controlStream, dataStream, err := c.openStream(ctx, addr) | 	controlStream, dataStream, err := c.openStream(ctx, addr) | ||||||
| 	if err != nil { | 	if err != nil { | ||||||
| 		log.Errorf("connect error, %s", err) | 		c.logger.Errorf("connect error, %s", err) | ||||||
| 		return err | 		return err | ||||||
| 	} | 	} | ||||||
| 
 | 
 | ||||||
|  | @ -91,7 +96,7 @@ func (c *Client) runBackground(ctx context.Context, addr string, controlStream C | ||||||
| 			var err error | 			var err error | ||||||
| 			controlStream, dataStream, err = c.openStream(ctx, addr) | 			controlStream, dataStream, err = c.openStream(ctx, addr) | ||||||
| 			if err != nil { | 			if err != nil { | ||||||
| 				log.Errorf("client reconnect error, %s", err) | 				c.logger.Errorf("client reconnect error, %s", err) | ||||||
| 				time.Sleep(time.Second) | 				time.Sleep(time.Second) | ||||||
| 				goto RECONNECT | 				goto RECONNECT | ||||||
| 			} | 			} | ||||||
|  | @ -110,7 +115,7 @@ func (c *Client) cleanStream(controlStream ClientControlStream, err error) { | ||||||
| 	errString := "" | 	errString := "" | ||||||
| 	if err != nil { | 	if err != nil { | ||||||
| 		errString = err.Error() | 		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.
 | 	// controlStream is nil represents that client is not connected.
 | ||||||
|  | @ -250,18 +255,18 @@ func (c *Client) handleFrame(f frame.Frame) { | ||||||
| 	switch ff := f.(type) { | 	switch ff := f.(type) { | ||||||
| 	case *frame.DataFrame: | 	case *frame.DataFrame: | ||||||
| 		if c.processor == nil { | 		if c.processor == nil { | ||||||
| 			log.Warnf("client processor has not been set") | 			c.logger.Warnf("client processor has not been set") | ||||||
| 		} else { | 		} else { | ||||||
| 			c.processor(ff) | 			c.processor(ff) | ||||||
| 		} | 		} | ||||||
| 	case *frame.BackFlowFrame: | 	case *frame.BackFlowFrame: | ||||||
| 		if c.receiver == nil { | 		if c.receiver == nil { | ||||||
| 			log.Warnf("client receiver has not been set") | 			c.logger.Warnf("client receiver has not been set") | ||||||
| 		} else { | 		} else { | ||||||
| 			c.receiver(ff) | 			c.receiver(ff) | ||||||
| 		} | 		} | ||||||
| 	default: | 	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.
 | // SetDataFrameObserver sets the data frame handler.
 | ||||||
| func (c *Client) SetDataFrameObserver(fn func(*frame.DataFrame)) { | func (c *Client) SetDataFrameObserver(fn func(*frame.DataFrame)) { | ||||||
| 	c.processor = fn | 	c.processor = fn | ||||||
| 	log.Debugf("SetDataFrameObserver") | 	c.logger.Debugf("SetDataFrameObserver") | ||||||
| } | } | ||||||
| 
 | 
 | ||||||
| // SetBackFlowFrameObserver sets the backflow frame handler.
 | // SetBackFlowFrameObserver sets the backflow frame handler.
 | ||||||
| func (c *Client) SetBackFlowFrameObserver(fn func(*frame.BackFlowFrame)) { | func (c *Client) SetBackFlowFrameObserver(fn func(*frame.BackFlowFrame)) { | ||||||
| 	c.receiver = fn | 	c.receiver = fn | ||||||
| 	log.Debugf("SetBackFlowFrameObserver") | 	c.logger.Debugf("SetBackFlowFrameObserver") | ||||||
| } | } | ||||||
| 
 | 
 | ||||||
| // SetObserveDataTags set the data tag list that will be observed.
 | // 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 { | func (c *Client) ClientId() string { | ||||||
| 	return c.clientId | 	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 | ||||||
|  | } | ||||||
|  |  | ||||||
		Loading…
	
		Reference in New Issue