mirror of
https://github.com/postmannen/ctrl.git
synced 2024-12-14 12:37:31 +00:00
removed procfunc type definition
This commit is contained in:
postmannen
parent
ea79bced4d
commit
13fd48843d
2 changed files with 81 additions and 90 deletions
61
process.go
61
process.go
|
|
@ -45,12 +45,33 @@ type process struct {
|
|||
processKind processKind
|
||||
// methodsAvailable
|
||||
methodsAvailable MethodsAvailable
|
||||
// Helper or service function that can do some kind of work
|
||||
// for the process.
|
||||
// The idea is that this can hold for example the map of the
|
||||
// the hello nodes to limit shared resources in the system as
|
||||
// a whole for sharing a map from the *server level.
|
||||
procFunc procFunc
|
||||
// procFunc is a function that will be started when a worker process
|
||||
// is started. If a procFunc is registered when creating a new process
|
||||
// the procFunc will be started as a go routine when the process is started,
|
||||
// and stopped when the process is stopped.
|
||||
//
|
||||
// A procFunc can be started both for publishing and subscriber processes.
|
||||
//
|
||||
// When used with a subscriber process the usecase is most likely to handle
|
||||
// some kind of state needed for a request type. The handlers themselves
|
||||
// can not hold state since they are only called once per message received,
|
||||
// and exits when the message is handled leaving no state behind. With a procfunc
|
||||
// we can have a process function running at all times tied to the process, and
|
||||
// this function can be able to hold the state needed in a certain scenario.
|
||||
//
|
||||
// With a subscriber handler you generally take the message in the handler and
|
||||
// pass it on to the procFunc by putting it on the procFuncCh<-, and the
|
||||
// message can then be read from the procFuncCh inside the procFunc, and we
|
||||
// can do some further work on it, for example update registry for metrics that
|
||||
// is needed for that specific request type.
|
||||
//
|
||||
// With a publisher process you can attach a static function that will do some
|
||||
// work to a request type, and publish the result.
|
||||
//
|
||||
// procFunc's can also be used to wrap in other types which we want to
|
||||
// work with. An example can be handling of metrics which the message
|
||||
// have no notion of, but a procFunc can have that wrapped in from when it was constructed.
|
||||
procFunc func(ctx context.Context, procFuncCh chan Message) error
|
||||
// The channel to send a messages to the procFunc go routine.
|
||||
// This is typically used within the methodHandler for so we
|
||||
// can pass messages between the procFunc and the handler.
|
||||
|
|
@ -110,34 +131,6 @@ func newProcess(ctx context.Context, metrics *metrics, natsConn *nats.Conn, proc
|
|||
return proc
|
||||
}
|
||||
|
||||
// procFunc is a function that will be started when a worker process
|
||||
// is started. If a procFunc is registered when creating a new process
|
||||
// the procFunc will be started as a go routine when the process is started,
|
||||
// and stopped when the process is stopped.
|
||||
//
|
||||
// A procFunc can be started both for publishing and subscriber processes.
|
||||
//
|
||||
// When used with a subscriber process the usecase is most likely to handle
|
||||
// some kind of state needed for a request type. The handlers themselves
|
||||
// can not hold state since they are only called once per message received,
|
||||
// and exits when the message is handled leaving no state behind. With a procfunc
|
||||
// we can have a process function running at all times tied to the process, and
|
||||
// this function can be able to hold the state needed in a certain scenario.
|
||||
//
|
||||
// With a subscriber handler you generally take the message in the handler and
|
||||
// pass it on to the procFunc by putting it on the procFuncCh<-, and the
|
||||
// message can then be read from the procFuncCh inside the procFunc, and we
|
||||
// can do some further work on it, for example update registry for metrics that
|
||||
// is needed for that specific request type.
|
||||
//
|
||||
// With a publisher process you can attach a static function that will do some
|
||||
// work to a request type, and publish the result.
|
||||
//
|
||||
// procFunc's can also be used to wrap in other types which we want to
|
||||
// work with. An example can be handling of metrics which the message
|
||||
// have no notion of, but a procFunc can have that wrapped in from when it was constructed.
|
||||
type procFunc func(ctx context.Context, procFuncCh chan Message) error
|
||||
|
||||
// The purpose of this function is to check if we should start a
|
||||
// publisher or subscriber process, where a process is a go routine
|
||||
// that will handle either sending or receiving messages on one
|
||||
|
|
|
|||
110
processes.go
110
processes.go
|
|
@ -254,42 +254,41 @@ func (s startup) pubREQHello(p process) {
|
|||
proc := newProcess(p.ctx, s.metrics, p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, processKindPublisher, nil, s.Signatures)
|
||||
|
||||
// Define the procFunc to be used for the process.
|
||||
proc.procFunc = procFunc(
|
||||
func(ctx context.Context, procFuncCh chan Message) error {
|
||||
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
|
||||
for {
|
||||
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
|
||||
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
|
||||
for {
|
||||
|
||||
d := fmt.Sprintf("Hello from %v\n", p.node)
|
||||
d := fmt.Sprintf("Hello from %v\n", p.node)
|
||||
|
||||
m := Message{
|
||||
FileName: "hello.log",
|
||||
Directory: "hello-messages",
|
||||
ToNode: Node(p.configuration.CentralNodeName),
|
||||
FromNode: Node(p.node),
|
||||
Data: []byte(d),
|
||||
Method: REQHello,
|
||||
ACKTimeout: 10,
|
||||
Retries: 1,
|
||||
}
|
||||
|
||||
sam, err := newSubjectAndMessage(m)
|
||||
if err != nil {
|
||||
// In theory the system should drop the message before it reaches here.
|
||||
p.processes.errorKernel.errSend(p, m, err)
|
||||
log.Printf("error: ProcessesStart: %v\n", err)
|
||||
}
|
||||
proc.toRingbufferCh <- []subjectAndMessage{sam}
|
||||
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-ctx.Done():
|
||||
er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name())
|
||||
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
|
||||
log.Printf("%v\n", er)
|
||||
return nil
|
||||
}
|
||||
m := Message{
|
||||
FileName: "hello.log",
|
||||
Directory: "hello-messages",
|
||||
ToNode: Node(p.configuration.CentralNodeName),
|
||||
FromNode: Node(p.node),
|
||||
Data: []byte(d),
|
||||
Method: REQHello,
|
||||
ACKTimeout: 10,
|
||||
Retries: 1,
|
||||
}
|
||||
})
|
||||
|
||||
sam, err := newSubjectAndMessage(m)
|
||||
if err != nil {
|
||||
// In theory the system should drop the message before it reaches here.
|
||||
p.processes.errorKernel.errSend(p, m, err)
|
||||
log.Printf("error: ProcessesStart: %v\n", err)
|
||||
}
|
||||
proc.toRingbufferCh <- []subjectAndMessage{sam}
|
||||
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-ctx.Done():
|
||||
er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name())
|
||||
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
|
||||
log.Printf("%v\n", er)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
go proc.spawnWorker(p.processes, p.natsConn)
|
||||
}
|
||||
|
||||
|
|
@ -351,32 +350,31 @@ func (s startup) subREQHello(p process) {
|
|||
// a handler are not able to hold state, and we need to hold the state
|
||||
// of the nodes we've received hello's from in the sayHelloNodes map,
|
||||
// which is the information we pass along to generate metrics.
|
||||
proc.procFunc = procFunc(
|
||||
func(ctx context.Context, procFuncCh chan Message) error {
|
||||
sayHelloNodes := make(map[Node]struct{})
|
||||
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
|
||||
sayHelloNodes := make(map[Node]struct{})
|
||||
|
||||
for {
|
||||
// Receive a copy of the message sent from the method handler.
|
||||
var m Message
|
||||
|
||||
select {
|
||||
case m = <-procFuncCh:
|
||||
case <-ctx.Done():
|
||||
er := fmt.Errorf("info: stopped handleFunc for: subscriber %v", proc.subject.name())
|
||||
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
|
||||
log.Printf("%v\n", er)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Add an entry for the node in the map
|
||||
sayHelloNodes[m.FromNode] = struct{}{}
|
||||
|
||||
// update the prometheus metrics
|
||||
s.metrics.promHelloNodesTotal.Set(float64(len(sayHelloNodes)))
|
||||
s.metrics.promHelloNodesContactLast.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime()
|
||||
for {
|
||||
// Receive a copy of the message sent from the method handler.
|
||||
var m Message
|
||||
|
||||
select {
|
||||
case m = <-procFuncCh:
|
||||
case <-ctx.Done():
|
||||
er := fmt.Errorf("info: stopped handleFunc for: subscriber %v", proc.subject.name())
|
||||
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
|
||||
log.Printf("%v\n", er)
|
||||
return nil
|
||||
}
|
||||
})
|
||||
|
||||
// Add an entry for the node in the map
|
||||
sayHelloNodes[m.FromNode] = struct{}{}
|
||||
|
||||
// update the prometheus metrics
|
||||
s.metrics.promHelloNodesTotal.Set(float64(len(sayHelloNodes)))
|
||||
s.metrics.promHelloNodesContactLast.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime()
|
||||
|
||||
}
|
||||
}
|
||||
go proc.spawnWorker(p.processes, p.natsConn)
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue