1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
ctrl/message_and_subject.go

164 lines
6.7 KiB
Go
Raw Normal View History

package steward
import (
"fmt"
"log"
"os"
)
// --- Message
type Message struct {
2021-12-29 07:29:11 +00:00
_ struct{} `cbor:",toarray"`
2021-08-25 14:17:33 +00:00
// The node to send the message to.
2021-06-29 06:21:42 +00:00
ToNode Node `json:"toNode" yaml:"toNode"`
2021-09-09 07:54:33 +00:00
// ToNodes to specify several hosts to send message to in the
// form of an slice/array.
2021-08-25 14:17:33 +00:00
ToNodes []Node `json:"toNodes,omitempty" yaml:"toNodes,omitempty"`
// The Unique ID of the message
ID int `json:"id" yaml:"id"`
2021-09-09 07:54:33 +00:00
// The actual data in the message. This is typically where we
// specify the cli commands to execute on a node, and this is
// also the field where we put the returned data in a reply
// message.
Data []byte `json:"data" yaml:"data"`
2021-09-09 07:54:33 +00:00
// Method, what request type to use, like REQCliCommand, REQHttpGet..
2021-04-13 11:50:00 +00:00
Method Method `json:"method" yaml:"method"`
// Additional arguments that might be needed when executing the
// method. Can be f.ex. an ip address if it is a tcp sender, or the
// shell command to execute in a cli session.
MethodArgs []string `json:"methodArgs" yaml:"methodArgs"`
2022-01-31 09:06:14 +00:00
// ArgSignature is the ed25519 signature of the methodArgs.
ArgSignature []byte `json:"argSignature" yaml:"argSignature"`
2021-04-13 11:50:00 +00:00
// ReplyMethod, is the method to use for the reply message.
// By default the reply method will be set to log to file, but
// you can override it setting your own here.
ReplyMethod Method `json:"replyMethod" yaml:"replyMethod"`
// Additional arguments that might be needed when executing the reply
// method. Can be f.ex. an ip address if it is a tcp sender, or the
// shell command to execute in a cli session.
ReplyMethodArgs []string `json:"replyMethodArgs" yaml:"replyMethodArgs"`
2021-09-22 14:08:55 +00:00
// IsReply are used to tell that this is a reply message. By default
// the system sends the output of a request method back to the node
// the message originated from. If it is a reply method we want the
// result of the reply message to be sent to the central server, so
// we can use this value if set to swap the toNode, and fromNode
// fields.
IsReply bool `json:"isReply" yaml:"isReply"`
2021-04-13 11:50:00 +00:00
// From what node the message originated
2022-01-26 08:23:02 +00:00
FromNode Node `json:"fromNode" yaml:"fromNode"`
// ACKTimeout for waiting for an ack message
ACKTimeout int `json:"ACKTimeout" yaml:"ACKTimeout"`
// RetryWait specified the time in seconds to wait between retries.
RetryWait int `json:"retryWait" yaml:"retryWait"`
// IsSubPublishedMsg enables timeout of publishing process, and is used together with process.isSubProcess to be able to terminate the sub processes publishers.
IsSubPublishedMsg bool `json:"isSubPublishedMsg" yaml:"isSubPublishedMsg"`
// Resend retries
Retries int `json:"retries" yaml:"retries"`
// The ACK timeout of the new message created via a request event.
ReplyACKTimeout int `json:"replyACKTimeout" yaml:"replyACKTimeout"`
2021-03-11 16:14:43 +00:00
// The retries of the new message created via a request event.
2021-04-13 12:37:17 +00:00
ReplyRetries int `json:"replyRetries" yaml:"replyRetries"`
2021-03-11 16:14:43 +00:00
// Timeout for long a process should be allowed to operate
MethodTimeout int `json:"methodTimeout" yaml:"methodTimeout"`
2021-09-16 10:37:46 +00:00
// Timeout for long a process should be allowed to operate
ReplyMethodTimeout int `json:"replyMethodTimeout" yaml:"replyMethodTimeout"`
2021-04-07 05:00:40 +00:00
// Directory is a string that can be used to create the
//directory structure when saving the result of some method.
// For example "syslog","metrics", or "metrics/mysensor"
// The type is typically used in the handler of a method.
Directory string `json:"directory" yaml:"directory"`
2021-09-09 07:54:33 +00:00
// FileName is used to be able to set a wanted name
2021-04-07 05:00:40 +00:00
// on a file being saved as the result of data being handled
// by a method handler.
FileName string `json:"fileName" yaml:"fileName"`
// PreviousMessage are used for example if a reply message is
// generated and we also need a copy of the details of the the
// initial request message.
PreviousMessage *Message
2022-06-20 09:17:23 +00:00
// Schedule
Schedule []int `json:"schedule" yaml:"schedule"`
// done is used to signal when a message is fully processed.
// This is used for signaling back to the ringbuffer that we are
// done with processing a message, and the message can be removed
// from the ringbuffer and into the time series log.
2021-03-11 16:14:43 +00:00
done chan struct{}
// ctx for the specifix message. Used for for example canceling
// scheduled messages.
// NB: Commented out this field for specific message context
// to be used within handlers, since it will override the structure
// we have today. Keeping the code for a bit incase it makes sense
// to implement later.
//ctx context.Context
}
// --- Subject
2021-08-16 11:01:12 +00:00
// Node is the type definition for the node who receive or send a message.
2021-06-29 06:21:42 +00:00
type Node string
// subject contains the representation of a subject to be used with one
// specific process
type Subject struct {
2021-08-16 11:01:12 +00:00
// node, the name of the node to receive the message.
ToNode string `json:"node" yaml:"toNode"`
// Event, event type like EventACK or EventNACK.
Event Event `json:"event" yaml:"event"`
// method, what is this message doing, etc. CLICommand, Syslog, etc.
Method Method `json:"method" yaml:"method"`
2021-03-09 06:43:55 +00:00
// messageCh is used by publisher kind processes to read new messages
// to be published. The content on this channel have been routed here
// from routeMessagesToPublish in *server.
// This channel is only used for publishing processes.
messageCh chan Message
}
// newSubject will return a new variable of the type subject, and insert
2022-06-07 09:52:30 +00:00
// all the values given as arguments. It will create the channel
// to receive new messages on the specific subject.
2022-06-07 09:52:30 +00:00
// The function will also verify that there is a methodHandler defined
// for the Request type.
func newSubject(method Method, node string) Subject {
// Get the Event type for the Method.
ma := method.GetMethodsAvailable()
mh, ok := ma.CheckIfExists(method)
//mh, ok := ma.Methodhandlers[method]
if !ok {
log.Printf("error: no Event type specified for the method: %v\n", method)
os.Exit(1)
}
return Subject{
ToNode: node,
Event: mh.getKind(),
Method: method,
messageCh: make(chan Message),
}
}
2022-06-07 09:52:30 +00:00
// newSubjectNoVerifyHandler will return a new variable of the type subject, and insert
// all the values given as arguments. It will create the channel
// to receive new messages on the specific subject.
// The function will not verify that there is a methodHandler defined
// for the Request type.
func newSubjectNoVerifyHandler(method Method, node string) Subject {
// Get the Event type for the Method.
return Subject{
ToNode: node,
Event: EventACK,
Method: method,
messageCh: make(chan Message),
}
}
// subjectName is the complete representation of a subject
type subjectName string
2021-08-16 11:01:12 +00:00
// Return a value of the subjectName for the subject as used with nats subject.
func (s Subject) name() subjectName {
return subjectName(fmt.Sprintf("%s.%s.%s", s.ToNode, s.Method, s.Event))
}