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

Compare commits

base: pavement:main
Branches Tags
pavement:main
pavement:v1.14
pavement:v1.12
pavement:v1.11
pavement:v1.10
pavement:v1.9
pavement:v1.8
pavement:v1.7
pavement:v1.6
pavement:v1.5
pavement:v1.4
pavement:v1.3
pavement:v.1.2
pavement:v.1.1
pavement:v1.0
..
compare: pavement:v1.3
Branches Tags
pavement:main
pavement:v1.14
pavement:v1.12
pavement:v1.11
pavement:v1.10
pavement:v1.9
pavement:v1.8
pavement:v1.7
pavement:v1.6
pavement:v1.5
pavement:v1.4
pavement:v1.3
pavement:v.1.2
pavement:v.1.1
pavement:v1.0

No commits in common. "main" and "v1.3" have entirely different histories.
main ... v1.3

10 changed files with 232 additions and 119 deletions
Show Stats Expand all files Collapse all files

51
ap.go
View File

@ -4,7 +4,6 @@ import (
"context"
"git.hpds.cc/Component/network"
"git.hpds.cc/Component/network/frame"
"git.hpds.cc/Component/network/log"
)
const (
@ -18,17 +17,15 @@ type AccessPoint interface {
// Connect to HPDS-Mq.
Connect() error
// SetDataTag will set the tag of data when invoking Write().
SetDataTag(tag frame.Tag)
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 frame.Tag, data []byte) error
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 frame.Tag, data []byte))
// Broadcast the data to all downstream
Broadcast(data []byte) error
SetReceiveHandler(fn func(tag byte, data []byte))
}
// hpds-AccessPoint
@ -36,8 +33,8 @@ type accessPoint struct {
name string
mqEndpoint string
client *network.Client
tag frame.Tag
fn func(frame.Tag, []byte)
tag uint8
fn func(byte, []byte)
}
var _ AccessPoint = &accessPoint{}
@ -45,7 +42,7 @@ var _ AccessPoint = &accessPoint{}
// NewAccessPoint create a hpds-AccessPoint
func NewAccessPoint(name string, opts ...Option) AccessPoint {
options := NewOptions(opts...)
client := network.NewClient(name, network.ClientTypeSource, options.ClientOptions...)
client := network.NewClient(name, network.ClientTypeProtocolGateway, options.ClientOptions...)
return &accessPoint{
name: name,
@ -56,25 +53,21 @@ func NewAccessPoint(name string, opts ...Option) AccessPoint {
// Write the data to downstream.
func (s *accessPoint) Write(data []byte) (int, error) {
err := s.WriteWithTag(s.tag, data)
if err != nil {
return 0, err
}
return len(data), nil
return len(data), s.WriteWithTag(s.tag, data)
}
// SetDataTag will set the tag of data when invoking Write().
func (s *accessPoint) SetDataTag(tag frame.Tag) {
func (s *accessPoint) SetDataTag(tag uint8) {
s.tag = tag
}
// Close will close the connection to MessageQueue.
// Close will close the connection to YoMo-MessageQueue.
func (s *accessPoint) Close() error {
if err := s.client.Close(); err != nil {
log.Errorf("%s Close(): %v", apLogPrefix, err)
s.client.Logger().Errorf("%sClose(): %v", apLogPrefix, err)
return err
}
log.Debugf("%s is closed", apLogPrefix)
s.client.Logger().Debugf("%s is closed", apLogPrefix)
return nil
}
@ -89,18 +82,18 @@ func (s *accessPoint) Connect() error {
err := s.client.Connect(context.Background(), s.mqEndpoint)
if err != nil {
log.Errorf("%s Connect() error: %s", apLogPrefix, err)
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 frame.Tag, data []byte) error {
func (s *accessPoint) WriteWithTag(tag uint8, data []byte) error {
f := frame.NewDataFrame()
f.SetCarriage(tag, data)
f.SetSourceId(s.client.ClientId())
log.Debugf("%s WriteWithTag: tid=%s, source_id=%s, data[%d]",
apLogPrefix, f.TransactionId(), f.SourceId(), len(data))
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)
}
@ -110,17 +103,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)) {
func (s *accessPoint) SetReceiveHandler(fn func(byte, []byte)) {
s.fn = fn
log.Debugf("%s SetReceiveHandler(%v)", apLogPrefix, s.fn)
}
// Broadcast Write the data to all downstream
func (s *accessPoint) Broadcast(data []byte) error {
f := frame.NewDataFrame()
f.SetCarriage(s.tag, data)
f.SetSourceId(s.client.ClientId())
f.SetBroadcast(true)
log.Debugf("Broadcast", "data_frame", f.String())
return s.client.WriteFrame(f)
s.client.Logger().Debugf("%sSetReceiveHandler(%v)", apLogPrefix, s.fn)
}

4
ap_test.go
View File

@ -7,9 +7,7 @@ import (
func TestAccessPointSendDataToServer(t *testing.T) {
ap := NewAccessPoint("test-ap")
defer func() {
_ = ap.Close()
}()
defer ap.Close()
// connect to server
err := ap.Connect()

33
go.mod
View File

@ -3,9 +3,9 @@ module git.hpds.cc/pavement/hpds_node
go 1.19
require (
git.hpds.cc/Component/network v0.0.0-20230421024959-bf7300c92a95
git.hpds.cc/Component/network v0.0.0-20221012021659-2433c68452d5
github.com/disintegration/imaging v1.6.2
github.com/quic-go/quic-go v0.33.0
github.com/lucas-clemente/quic-go v0.29.1
github.com/stretchr/testify v1.8.0
github.com/u2takey/ffmpeg-go v0.4.1
gopkg.in/yaml.v3 v3.0.1
@ -15,31 +15,30 @@ require (
git.hpds.cc/Component/mq_coder v0.0.0-20221010064749-174ae7ae3340 // indirect
github.com/aws/aws-sdk-go v1.38.20 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/fsnotify/fsnotify v1.4.9 // indirect
github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
github.com/golang/mock v1.6.0 // indirect
github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/kr/text v0.2.0 // indirect
github.com/marten-seemann/qtls-go1-18 v0.1.2 // indirect
github.com/marten-seemann/qtls-go1-19 v0.1.0 // indirect
github.com/matoous/go-nanoid/v2 v2.0.0 // indirect
github.com/onsi/ginkgo/v2 v2.2.0 // indirect
github.com/nxadm/tail v1.4.8 // indirect
github.com/onsi/ginkgo v1.16.4 // indirect
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/rogpeppe/go-internal v1.9.0 // indirect
github.com/u2takey/go-utils v0.3.1 // indirect
go.uber.org/atomic v1.7.0 // indirect
go.uber.org/multierr v1.6.0 // indirect
go.uber.org/zap v1.23.0 // indirect
golang.org/x/crypto v0.4.0 // indirect
golang.org/x/exp v0.0.0-20221205204356-47842c84f3db // indirect
golang.org/x/image v0.1.0 // indirect
golang.org/x/mod v0.6.0 // indirect
golang.org/x/net v0.5.0 // indirect
golang.org/x/sys v0.4.0 // indirect
golang.org/x/tools v0.2.0 // indirect
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect
golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8 // indirect
golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e // indirect
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
golang.org/x/tools v0.1.10 // indirect
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
)
//replace (
// git.hpds.cc/Component/network => ../network
//)

30
metadata.go Normal file
View File

@ -0,0 +1,30 @@
package hpds_node
import (
"git.hpds.cc/Component/network"
"git.hpds.cc/Component/network/frame"
)
type metadata struct{}
func (m *metadata) Encode() []byte {
return nil
}
type metadataBuilder struct {
m *metadata
}
func newMetadataBuilder() network.MetadataBuilder {
return &metadataBuilder{
m: &metadata{},
}
}
func (builder *metadataBuilder) Build(f *frame.HandshakeFrame) (network.Metadata, error) {
return builder.m, nil
}
func (builder *metadataBuilder) Decode(buf []byte) (network.Metadata, error) {
return builder.m, nil
}

39
mq.go
View File

@ -4,8 +4,6 @@ import (
"context"
"encoding/json"
"fmt"
"git.hpds.cc/Component/network/metadata"
"git.hpds.cc/Component/network/router"
"net/http"
"git.hpds.cc/Component/network"
@ -70,10 +68,10 @@ var _ MessageQueue = &messageQueue{}
// NewMqWithOptions create a messageQueue instance.
func NewMqWithOptions(name string, opts ...Option) MessageQueue {
options := NewOptions(opts...)
emitter := createMessageQueueServer(name, options, nil)
_ = emitter.ConfigMesh(options.MeshConfigURL)
zipper := createMessageQueueServer(name, options, nil)
_ = zipper.ConfigMesh(options.MeshConfigURL)
return emitter
return zipper
}
// NewMq create a messageQueue instance from config files.
@ -88,17 +86,17 @@ func NewMq(conf string) (MessageQueue, error) {
options := NewOptions()
options.MqAddr = listenAddr
emitter := createMessageQueueServer(confWf.Name, options, confWf)
zipper := createMessageQueueServer(confWf.Name, options, confWf)
// messageQueue workflow
err = emitter.configWorkflow(confWf)
err = zipper.configWorkflow(confWf)
return emitter, err
return zipper, err
}
// NewDownstreamMq create a messageQueue descriptor for downstream messageQueue.
func NewDownstreamMq(name string, opts ...Option) MessageQueue {
options := NewOptions(opts...)
client := network.NewClient(name, network.ClientTypeUpstreamEmitter, options.ClientOptions...)
client := network.NewClient(name, network.ClientTypeMessageQueue, options.ClientOptions...)
return &messageQueue{
name: name,
@ -122,9 +120,6 @@ func createMessageQueueServer(name string, options *Options, cfg *config.Workflo
z.init()
return z
}
func (z *messageQueue) Logger() log.Logger {
return log.Default()
}
// ConfigWorkflow will read workflows from config files and register them to messageQueue.
func (z *messageQueue) ConfigWorkflow(conf string) error {
@ -139,12 +134,8 @@ func (z *messageQueue) ConfigWorkflow(conf string) error {
func (z *messageQueue) configWorkflow(config *config.WorkflowConfig) error {
z.wfc = config
z.server.ConfigMetadataBuilder(metadata.DefaultBuilder())
funcList := make([]string, len(config.Functions))
for k, v := range config.Functions {
funcList[k] = v.Name
}
z.server.ConfigRouter(router.Default(funcList))
z.server.ConfigMetadataBuilder(newMetadataBuilder())
z.server.ConfigRouter(newRouter(config.Functions))
return nil
}
@ -197,11 +188,11 @@ func (z *messageQueue) ListenAndServe() error {
log.Debugf("%sCreating MessageQueue Server ...", mqLogPrefix)
// check downstream zippers
for _, ds := range z.downstreamMqs {
if dsEmitter, ok := ds.(*messageQueue); ok {
go func(dsEmitter *messageQueue) {
_ = dsEmitter.client.Connect(context.Background(), dsEmitter.addr)
z.server.AddDownstreamServer(dsEmitter.addr, dsEmitter.client)
}(dsEmitter)
if dsZipper, ok := ds.(*messageQueue); ok {
go func(dsZipper *messageQueue) {
_ = dsZipper.client.Connect(context.Background(), dsZipper.addr)
z.server.AddDownstreamServer(dsZipper.addr, dsZipper.client)
}(dsZipper)
}
}
return z.server.ListenAndServe(context.Background(), z.addr)
@ -262,7 +253,7 @@ func (z *messageQueue) Stats() int {
log.Printf("[%s] all downstream mq connected: %d", z.name, len(z.server.DownStreams()))
for k, v := range z.server.DownStreams() {
log.Printf("[%s] |> [%s] %v", z.name, k, v)
log.Printf("[%s] |> [%s] %s", z.name, k, v.ServerAddr())
}
log.Printf("[%s] total DataFrames received: %d", z.name, z.server.StatsCounter())

9
mq_notwindows.go
View File

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

4
mq_windows.go
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.Infof("%s Listening SIGTERM/SIGINT...", mqLogPrefix)
log.Printf("%sListening SIGTERM/SIGINT...", mqLogPrefix)
for p1 := range c {
log.Printf("Received signal: %s", p1)
if p1 == syscall.SIGTERM || p1 == syscall.SIGINT {
log.Infof("%s graceful shutting down ... %s", mqLogPrefix, p1)
log.Printf("graceful shutting down ... %s", p1)
os.Exit(0)
}
}

16
options.go
View File

@ -2,10 +2,10 @@ package hpds_node
import (
"crypto/tls"
"github.com/lucas-clemente/quic-go"
"git.hpds.cc/Component/network"
"git.hpds.cc/Component/network/frame"
"git.hpds.cc/Component/network/log"
"github.com/quic-go/quic-go"
)
const (
@ -101,7 +101,7 @@ func WithCredential(payload string) Option {
}
// WithObserveDataTags sets client data tag list.
func WithObserveDataTags(tags ...frame.Tag) Option {
func WithObserveDataTags(tags ...byte) Option {
return func(o *Options) {
o.ClientOptions = append(
o.ClientOptions,
@ -110,6 +110,16 @@ func WithObserveDataTags(tags ...frame.Tag) Option {
}
}
// WithLogger sets the client logger
func WithLogger(logger log.Logger) Option {
return func(o *Options) {
o.ClientOptions = append(
o.ClientOptions,
network.WithLogger(logger),
)
}
}
// NewOptions creates a new options for YoMo-Client.
func NewOptions(opts ...Option) *Options {
options := &Options{}

103
router.go Normal file
View File

@ -0,0 +1,103 @@
package hpds_node
import (
"git.hpds.cc/Component/network"
"git.hpds.cc/pavement/hpds_node/config"
"sync"
)
type router struct {
r *route
}
func newRouter(functions []config.App) network.Router {
return &router{r: newRoute(functions)}
}
func (r *router) Route(metadata network.Metadata) network.Route {
return r.r
}
func (r *router) Clean() {
r.r = nil
}
type route struct {
functions []config.App
data map[byte]map[string]string
mu sync.RWMutex
}
func newRoute(functions []config.App) *route {
return &route{
functions: functions,
data: make(map[byte]map[string]string),
}
}
func (r *route) Add(connId string, name string, observeDataTags []byte) (err error) {
r.mu.Lock()
defer r.mu.Unlock()
ok := false
for _, v := range r.functions {
if v.Name == name {
ok = true
break
}
}
if !ok {
//return fmt.Errorf("SFN[%s] does not exist in config functions", name)
//如果不存在，自动增加
item := config.App{
Name: name,
}
r.functions = append(r.functions, item)
}
//去除只能订阅一次的问题
//LOOP:
//for _, connects := range r.data {
// for connKey, n := range connects {
// if n == name {
// err = hpds_err.NewDuplicateNameError(connKey, fmt.Errorf("node:Stream Function[%s] is already linked to another connection", name))
// delete(connects, connKey)
// break LOOP
// }
// }
//}
for _, tag := range observeDataTags {
connects := r.data[tag]
if connects == nil {
connects = make(map[string]string)
r.data[tag] = connects
}
r.data[tag][connId] = name
}
return err
}
func (r *route) Remove(connId string) error {
r.mu.Lock()
defer r.mu.Unlock()
for _, connects := range r.data {
delete(connects, connId)
}
return nil
}
func (r *route) GetForwardRoutes(tag byte) []string {
r.mu.RLock()
defer r.mu.RUnlock()
var keys []string
if connects := r.data[tag]; connects != nil {
for k := range connects {
keys = append(keys, k)
}
}
return keys
}

62
stream_func.go
View File

@ -14,7 +14,7 @@ const (
type StreamFunction interface {
// SetObserveDataTags set the data tag list that will be observed
// Deprecated: use hpds.WithObserveDataTags instead
SetObserveDataTags(tag ...frame.Tag)
SetObserveDataTags(tag ...byte)
// SetHandler set the handler function, which accept the raw bytes data and return the tag & response
SetHandler(fn network.AsyncHandler) error
// SetErrorHandler set the error handler function when server error occurs
@ -26,7 +26,7 @@ type StreamFunction interface {
// Close will close the connection
Close() error
// Write Send a data to mq.
Write(tag frame.Tag, carriage []byte) error
Write(tag byte, carriage []byte) error
}
// NewStreamFunction create a stream function.
@ -35,9 +35,9 @@ func NewStreamFunction(name string, opts ...Option) StreamFunction {
client := network.NewClient(name, network.ClientTypeStreamFunction, options.ClientOptions...)
sf := &streamFunction{
name: name,
emitterEndpoint: options.MqAddr,
zipperEndpoint: options.MqAddr,
client: client,
observeDataTags: make([]frame.Tag, 0),
observeDataTags: make([]byte, 0),
}
return sf
@ -48,9 +48,9 @@ var _ StreamFunction = &streamFunction{}
// streamFunction implements StreamFunction interface.
type streamFunction struct {
name string
emitterEndpoint string
zipperEndpoint string
client *network.Client
observeDataTags []frame.Tag // tag list that will be observed
observeDataTags []byte // tag list that will be observed
fn network.AsyncHandler // user's function which will be invoked when data arrived
pfn network.PipeHandler
pIn chan []byte
@ -59,21 +59,21 @@ type streamFunction struct {
// SetObserveDataTags set the data tag list that will be observed.
// Deprecated: use hpds.WithObserveDataTags instead
func (s *streamFunction) SetObserveDataTags(tag ...frame.Tag) {
func (s *streamFunction) SetObserveDataTags(tag ...byte) {
s.client.SetObserveDataTags(tag...)
s.client.Logger().Debugf("%s SetObserveDataTag(%v)", streamFunctionLogPrefix, s.observeDataTags)
s.client.Logger().Debugf("%sSetObserveDataTag(%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("%s SetHandler(%v)", streamFunctionLogPrefix, s.fn)
s.client.Logger().Debugf("%sSetHandler(%v)", streamFunctionLogPrefix, s.fn)
return nil
}
func (s *streamFunction) SetPipeHandler(fn network.PipeHandler) error {
s.pfn = fn
s.client.Logger().Debugf("%s SetHandler(%v)", streamFunctionLogPrefix, s.fn)
s.client.Logger().Debugf("%sSetHandler(%v)", streamFunctionLogPrefix, s.fn)
return nil
}
@ -83,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("%s receive DataFrame, tag=%# x, carriage length=%d", streamFunctionLogPrefix, data.Tag(), len(data.GetCarriage()))
s.client.Logger().Debugf("%sreceive DataFrame, tag=%# x, carriage=%# x", streamFunctionLogPrefix, data.Tag(), data.GetCarriage())
s.onDataFrame(data.GetCarriage(), data.GetMetaFrame())
})
@ -101,19 +101,19 @@ func (s *streamFunction) Connect() error {
for {
data := <-s.pOut
if data != nil {
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)
_ = s.client.WriteFrame(frm)
s.client.Logger().Debugf("%spipe fn send: tag=%#x, data=%# x", streamFunctionLogPrefix, data.Tag, data.Carriage)
frame := frame.NewDataFrame()
// todo: frame.SetTransactionId
frame.SetCarriage(data.Tag, data.Carriage)
s.client.WriteFrame(frame)
}
}
}()
}
err := s.client.Connect(context.Background(), s.emitterEndpoint)
err := s.client.Connect(context.Background(), s.zipperEndpoint)
if err != nil {
s.client.Logger().Errorf("%s Connect() error: %s", streamFunctionLogPrefix, err)
s.client.Logger().Errorf("%sConnect() error: %s", streamFunctionLogPrefix, err)
}
return err
}
@ -130,7 +130,7 @@ func (s *streamFunction) Close() error {
if s.client != nil {
if err := s.client.Close(); err != nil {
s.client.Logger().Errorf("%s Close(): %v", err)
s.client.Logger().Errorf("%sClose(): %v", err)
return err
}
}
@ -140,38 +140,38 @@ 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("%s onDataFrame ->[%s]", streamFunctionLogPrefix, s.name)
s.client.Logger().Infof("%sonDataFrame ->[%s]", streamFunctionLogPrefix, s.name)
if s.fn != nil {
go func() {
s.client.Logger().Debugf("%s execute-start fn: data[%d]", streamFunctionLogPrefix, len(data)) //, frame.Shortly(data)
s.client.Logger().Debugf("%sexecute-start fn: data[%d]=%# x", streamFunctionLogPrefix, len(data), frame.Shortly(data))
// invoke serverless
tag, resp := s.fn(data)
s.client.Logger().Debugf("%s execute-done fn: tag=%#x, resp[%d]", streamFunctionLogPrefix, tag, len(resp))
s.client.Logger().Debugf("%sexecute-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("%s start WriteFrame(): tag=%#x, data[%d]", streamFunctionLogPrefix, tag, len(resp))
s.client.Logger().Debugf("%sstart 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()
frame := frame.NewDataFrame()
// reuse transactionId
frm.SetTransactionId(metaFrame.TransactionId())
frame.SetTransactionId(metaFrame.TransactionId())
// reuse sourceId
frm.SetSourceId(metaFrame.SourceId())
frm.SetCarriage(tag, resp)
_ = s.client.WriteFrame(frm)
frame.SetSourceId(metaFrame.SourceId())
frame.SetCarriage(tag, resp)
s.client.WriteFrame(frame)
}
}()
} else if s.pfn != nil {
s.client.Logger().Debugf("%s pipe fn receive: data[%d]", streamFunctionLogPrefix, len(data))
s.client.Logger().Debugf("%spipe fn receive: data[%d]=%# x", streamFunctionLogPrefix, len(data), data)
s.pIn <- data
} else {
s.client.Logger().Warnf("%s StreamFunction is nil", streamFunctionLogPrefix)
s.client.Logger().Warnf("%sStreamFunction is nil", streamFunctionLogPrefix)
}
}
// Send a DataFrame to mq.
func (s *streamFunction) Write(tag frame.Tag, carriage []byte) error {
func (s *streamFunction) Write(tag byte, carriage []byte) error {
fm := frame.NewDataFrame()
fm.SetCarriage(tag, carriage)
return s.client.WriteFrame(fm)