package hpds_node import ( "context" "git.hpds.cc/Component/network" "git.hpds.cc/Component/network/frame" ) const ( apLogPrefix = "\033[32m[hpds:access_point]\033[0m " ) // AccessPoint is responsible for sending data to hpds. type AccessPoint interface { // Close will close the connection to HPDS-Mq. Close() error // Connect to HPDS-Mq. Connect() error // SetDataTag will set the tag of data when invoking Write(). SetDataTag(tag uint8) // Write the data to downstream. Write(p []byte) (n int, err error) // WriteWithTag will write data with specified tag, default transactionId is epoch time. WriteWithTag(tag uint8, data []byte) error // SetErrorHandler set the error handler function when server error occurs SetErrorHandler(fn func(err error)) // SetReceiveHandler [Experimental] set to observe handler function SetReceiveHandler(fn func(tag byte, data []byte)) } // hpds-AccessPoint type accessPoint struct { name string mqEndpoint string client *network.Client tag uint8 fn func(byte, []byte) } var _ AccessPoint = &accessPoint{} // NewAccessPoint create a hpds-AccessPoint func NewAccessPoint(name string, opts ...Option) AccessPoint { options := NewOptions(opts...) client := network.NewClient(name, network.ClientTypeProtocolGateway, options.ClientOptions...) return &accessPoint{ name: name, mqEndpoint: options.MqAddr, client: client, } } // Write the data to downstream. func (s *accessPoint) Write(data []byte) (int, error) { return len(data), s.WriteWithTag(s.tag, data) } // SetDataTag will set the tag of data when invoking Write(). func (s *accessPoint) SetDataTag(tag uint8) { s.tag = tag } // Close will close the connection to MessageQueue. func (s *accessPoint) Close() error { if err := s.client.Close(); err != nil { s.client.Logger().Errorf("%sClose(): %v", apLogPrefix, err) return err } s.client.Logger().Debugf("%s is closed", apLogPrefix) return nil } // Connect to YoMo-MessageQueue. func (s *accessPoint) Connect() error { // set backFlowFrame handler s.client.SetBackFlowFrameObserver(func(frm *frame.BackFlowFrame) { if s.fn != nil { s.fn(frm.GetDataTag(), frm.GetCarriage()) } }) err := s.client.Connect(context.Background(), s.mqEndpoint) if err != nil { s.client.Logger().Errorf("%sConnect() error: %s", apLogPrefix, err) } return err } // WriteWithTag will write data with specified tag, default transactionID is epoch time. func (s *accessPoint) WriteWithTag(tag uint8, data []byte) error { f := frame.NewDataFrame() f.SetCarriage(tag, data) f.SetSourceId(s.client.ClientId()) s.client.Logger().Debugf("%sWriteWithTag: tid=%s, source_id=%s, data[%d]=%# x", apLogPrefix, f.TransactionId(), f.SourceId(), len(data), frame.Shortly(data)) return s.client.WriteFrame(f) } // SetErrorHandler set the error handler function when server error occurs func (s *accessPoint) SetErrorHandler(fn func(err error)) { s.client.SetErrorHandler(fn) } // SetReceiveHandler [Experimental] set to observe handler function func (s *accessPoint) SetReceiveHandler(fn func(byte, []byte)) { s.fn = fn s.client.Logger().Debugf("%sSetReceiveHandler(%v)", apLogPrefix, s.fn) }