pavement
/
hpds_node
3
0
Fork 0
Code Issues Pull Requests Projects Releases 14 Wiki Activity
Browse Source

1、修改bug

tags/v1.10 v1.10
wangjian 1 year ago
parent
2c72ebec4f
commit
be7c3e11c6
5 changed files with 25 additions and 28 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 8
      ap.go
  2. 4
      go.mod
  3. 8
      mq_notwindows.go
  4. 4
      mq_windows.go
  5. 29
      stream_func.go

8
ap.go
Unescape View File

@ -71,7 +71,7 @@ func (s *accessPoint) SetDataTag(tag frame.Tag) {
// Close will close the connection to MessageQueue.
func (s *accessPoint) Close() error {
if err := s.client.Close(); err != nil {
log.Errorf("%sClose(): %v", apLogPrefix, err)
log.Errorf("%s Close(): %v", apLogPrefix, err)
return err
}
log.Debugf("%s is closed", apLogPrefix)
@ -89,7 +89,7 @@ func (s *accessPoint) Connect() error {
err := s.client.Connect(context.Background(), s.mqEndpoint)
if err != nil {
log.Errorf("%sConnect() error: %s", apLogPrefix, err)
log.Errorf("%s Connect() error: %s", apLogPrefix, err)
}
return err
}
@ -99,7 +99,7 @@ func (s *accessPoint) WriteWithTag(tag frame.Tag, data []byte) error {
f := frame.NewDataFrame()
f.SetCarriage(tag, data)
f.SetSourceId(s.client.ClientId())
log.Debugf("%sWriteWithTag: tid=%s, source_id=%s, data[%d]=%# x",
log.Debugf("%s WriteWithTag: tid=%s, source_id=%s, data[%d]=%# x",
apLogPrefix, f.TransactionId(), f.SourceId(), len(data), frame.Shortly(data))
return s.client.WriteFrame(f)
}
@ -112,7 +112,7 @@ func (s *accessPoint) SetErrorHandler(fn func(err error)) {
// SetReceiveHandler [Experimental] set to observe handler function
func (s *accessPoint) SetReceiveHandler(fn func(frame.Tag, []byte)) {
s.fn = fn
log.Debugf("%sSetReceiveHandler(%v)", apLogPrefix, s.fn)
log.Debugf("%s SetReceiveHandler(%v)", apLogPrefix, s.fn)
}
// Broadcast Write the data to all downstream

4
go.mod
Unescape View File

@ -3,8 +3,9 @@ module git.hpds.cc/pavement/hpds_node
go 1.19
require (
git.hpds.cc/Component/network v0.0.0-20230405091029-b109c53056fb
git.hpds.cc/Component/network v0.0.0-20230405125523-abb93c020ff5
github.com/disintegration/imaging v1.6.2
github.com/quic-go/quic-go v0.33.0
github.com/stretchr/testify v1.8.0
github.com/u2takey/ffmpeg-go v0.4.1
gopkg.in/yaml.v3 v3.0.1
@ -24,7 +25,6 @@ require (
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/quic-go/qtls-go1-19 v0.2.1 // indirect
github.com/quic-go/qtls-go1-20 v0.1.1 // indirect
github.com/quic-go/quic-go v0.33.0 // indirect
github.com/rogpeppe/go-internal v1.9.0 // indirect
github.com/u2takey/go-utils v0.3.1 // indirect
go.uber.org/atomic v1.7.0 // indirect

8
mq_notwindows.go
Unescape View File

@ -21,11 +21,11 @@ func (z *messageQueue) init() {
go func() {
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGTERM, syscall.SIGUSR2, syscall.SIGUSR1, syscall.SIGINT)
log.Printf("%sListening SIGUSR1, SIGUSR2, SIGTERM/SIGINT...", mqLogPrefix)
log.Infof("%sListening SIGUSR1, SIGUSR2, SIGTERM/SIGINT...", mqLogPrefix)
for p1 := range c {
log.Printf("Received signal: %s", p1)
log.Infof("Received signal: %s", p1)
if p1 == syscall.SIGTERM || p1 == syscall.SIGINT {
log.Printf("graceful shutting down ... %s", p1)
log.Infof("graceful shutting down ... %s", p1)
_ = z.Close()
os.Exit(0)
} else if p1 == syscall.SIGUSR2 {
@ -33,7 +33,7 @@ func (z *messageQueue) init() {
runtime.ReadMemStats(&m)
fmt.Printf("\tNumGC = %v\n", m.NumGC)
} else if p1 == syscall.SIGUSR1 {
log.Printf("print MessageQueue stats(): %d", z.Stats())
log.Infof("print MessageQueue stats(): %d", z.Stats())
}
}
}()

4
mq_windows.go
Unescape View File

@ -13,11 +13,11 @@ func (z *messageQueue) init() {
go func() {
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGTERM, syscall.SIGINT)
log.Printf("%sListening SIGTERM/SIGINT...", mqLogPrefix)
log.Infof("%s Listening SIGTERM/SIGINT...", mqLogPrefix)
for p1 := range c {
log.Printf("Received signal: %s", p1)
if p1 == syscall.SIGTERM || p1 == syscall.SIGINT {
log.Printf("graceful shutting down ... %s", p1)
log.Infof("%s graceful shutting down ... %s", mqLogPrefix, p1)
os.Exit(0)
}
}

29
stream_func.go
Unescape View File

@ -61,22 +61,19 @@ type streamFunction struct {
// Deprecated: use hpds.WithObserveDataTags instead
func (s *streamFunction) SetObserveDataTags(tag ...frame.Tag) {
s.client.SetObserveDataTags(tag...)
//co := network.WithObserveDataTags(tag...)
//s.client.
//s.client.SetObserveDataTags(tag...)
s.client.Logger().Debugf("%sSetObserveDataTag(%v)", streamFunctionLogPrefix, s.observeDataTags)
s.client.Logger().Debugf("%s SetObserveDataTag(%v)", streamFunctionLogPrefix, s.observeDataTags)
}
// SetHandler set the handler function, which accept the raw bytes data and return the tag & response.
func (s *streamFunction) SetHandler(fn network.AsyncHandler) error {
s.fn = fn
s.client.Logger().Debugf("%sSetHandler(%v)", streamFunctionLogPrefix, s.fn)
s.client.Logger().Debugf("%s SetHandler(%v)", streamFunctionLogPrefix, s.fn)
return nil
}
func (s *streamFunction) SetPipeHandler(fn network.PipeHandler) error {
s.pfn = fn
s.client.Logger().Debugf("%sSetHandler(%v)", streamFunctionLogPrefix, s.fn)
s.client.Logger().Debugf("%s SetHandler(%v)", streamFunctionLogPrefix, s.fn)
return nil
}
@ -86,7 +83,7 @@ func (s *streamFunction) Connect() error {
s.client.Logger().Debugf("%s Connect()", streamFunctionLogPrefix)
// notify underlying network operations, when data with tag we observed arrived, invoke the func
s.client.SetDataFrameObserver(func(data *frame.DataFrame) {
s.client.Logger().Debugf("%sreceive DataFrame, tag=%# x, carriage=%# x", streamFunctionLogPrefix, data.Tag(), data.GetCarriage())
s.client.Logger().Debugf("%s receive DataFrame, tag=%# x, carriage=%# x", streamFunctionLogPrefix, data.Tag(), data.GetCarriage())
s.onDataFrame(data.GetCarriage(), data.GetMetaFrame())
})
@ -104,7 +101,7 @@ func (s *streamFunction) Connect() error {
for {
data := <-s.pOut
if data != nil {
s.client.Logger().Debugf("%spipe fn send: tag=%#x, data=%# x", streamFunctionLogPrefix, data.Tag, data.Carriage)
s.client.Logger().Debugf("%s pipe fn send: tag=%#x, data=%# x", streamFunctionLogPrefix, data.Tag, data.Carriage)
frm := frame.NewDataFrame()
// todo: frm.SetTransactionId
frm.SetCarriage(data.Tag, data.Carriage)
@ -116,7 +113,7 @@ func (s *streamFunction) Connect() error {
err := s.client.Connect(context.Background(), s.emitterEndpoint)
if err != nil {
s.client.Logger().Errorf("%sConnect() error: %s", streamFunctionLogPrefix, err)
s.client.Logger().Errorf("%s Connect() error: %s", streamFunctionLogPrefix, err)
}
return err
}
@ -133,7 +130,7 @@ func (s *streamFunction) Close() error {
if s.client != nil {
if err := s.client.Close(); err != nil {
s.client.Logger().Errorf("%sClose(): %v", err)
s.client.Logger().Errorf("%s Close(): %v", err)
return err
}
}
@ -143,17 +140,17 @@ func (s *streamFunction) Close() error {
// when DataFrame we observed arrived, invoke the user's function
func (s *streamFunction) onDataFrame(data []byte, metaFrame *frame.MetaFrame) {
s.client.Logger().Infof("%sonDataFrame ->[%s]", streamFunctionLogPrefix, s.name)
s.client.Logger().Infof("%s onDataFrame ->[%s]", streamFunctionLogPrefix, s.name)
if s.fn != nil {
go func() {
s.client.Logger().Debugf("%sexecute-start fn: data[%d]=%# x", streamFunctionLogPrefix, len(data), frame.Shortly(data))
s.client.Logger().Debugf("%s execute-start fn: data[%d]=%# x", streamFunctionLogPrefix, len(data), frame.Shortly(data))
// invoke serverless
tag, resp := s.fn(data)
s.client.Logger().Debugf("%sexecute-done fn: tag=%#x, resp[%d]=%# x", streamFunctionLogPrefix, tag, len(resp), frame.Shortly(resp))
s.client.Logger().Debugf("%s execute-done fn: tag=%#x, resp[%d]=%# x", streamFunctionLogPrefix, tag, len(resp), frame.Shortly(resp))
// if resp is not nil, means the user's function has returned something, we should send it to the mq
if len(resp) != 0 {
s.client.Logger().Debugf("%sstart WriteFrame(): tag=%#x, data[%d]=%# x", streamFunctionLogPrefix, tag, len(resp), frame.Shortly(resp))
s.client.Logger().Debugf("%s start WriteFrame(): tag=%#x, data[%d]=%# x", streamFunctionLogPrefix, tag, len(resp), frame.Shortly(resp))
// build a DataFrame
// TODO: seems we should implement a DeepCopy() of MetaFrame in the future
frm := frame.NewDataFrame()
@ -166,10 +163,10 @@ func (s *streamFunction) onDataFrame(data []byte, metaFrame *frame.MetaFrame) {
}
}()
} else if s.pfn != nil {
s.client.Logger().Debugf("%spipe fn receive: data[%d]=%# x", streamFunctionLogPrefix, len(data), data)
s.client.Logger().Debugf("%s pipe fn receive: data[%d]=%# x", streamFunctionLogPrefix, len(data), data)
s.pIn <- data
} else {
s.client.Logger().Warnf("%sStreamFunction is nil", streamFunctionLogPrefix)
s.client.Logger().Warnf("%s StreamFunction is nil", streamFunctionLogPrefix)
}
}

Write Preview
Loading…
Cancel
Save