2021-01-28 10:17:54 +00:00
// Notes:
2021-02-01 10:13:38 +00:00
package steward
2021-01-25 14:23:00 +00:00
import (
2021-07-02 09:26:52 +00:00
"context"
2021-01-25 14:23:00 +00:00
"fmt"
"log"
2021-03-30 08:37:16 +00:00
"net"
2021-08-23 10:47:33 +00:00
"net/http"
2021-03-02 12:46:02 +00:00
"os"
2021-05-12 07:50:03 +00:00
"path/filepath"
2021-01-25 14:23:00 +00:00
"time"
"github.com/nats-io/nats.go"
2021-09-10 08:21:33 +00:00
"github.com/prometheus/client_golang/prometheus"
2021-01-25 14:23:00 +00:00
)
2021-02-26 06:55:28 +00:00
type processName string
2021-03-09 03:55:51 +00:00
// Will return a process name made up of subjectName+processKind
2021-02-26 06:55:28 +00:00
func processNameGet ( sn subjectName , pk processKind ) processName {
pn := fmt . Sprintf ( "%s_%s" , sn , pk )
return processName ( pn )
}
2021-01-28 10:17:54 +00:00
// server is the structure that will hold the state about spawned
// processes on a local instance.
2021-01-27 13:02:57 +00:00
type server struct {
2021-07-02 09:26:52 +00:00
// The main background context
ctx context . Context
// The CancelFunc for the main context
2021-08-11 10:23:37 +00:00
cancel context . CancelFunc
2021-03-01 19:49:43 +00:00
// Configuration options used for running the server
configuration * Configuration
// The nats connection to the broker
2021-01-27 13:02:57 +00:00
natsConn * nats . Conn
2021-07-02 06:38:44 +00:00
// net listener for communicating via the steward socket
2021-08-09 12:41:31 +00:00
StewardSocket net . Listener
2021-03-03 14:44:32 +00:00
// processes holds all the information about running processes
processes * processes
2021-02-04 12:26:10 +00:00
// The name of the node
nodeName string
2022-04-01 05:09:55 +00:00
// toRingBufferCh are the channel where new messages in a bulk
2022-01-20 07:17:37 +00:00
// format (slice) are put into the system.
2021-11-09 12:18:58 +00:00
//
// In general the ringbuffer will read this
2022-01-20 07:17:37 +00:00
// channel, unfold each slice, and put single messages on the buffer.
2022-04-01 05:09:55 +00:00
toRingBufferCh chan [ ] subjectAndMessage
2022-06-20 12:22:42 +00:00
// directSAMSCh
directSAMSCh chan [ ] subjectAndMessage
2021-02-24 09:58:02 +00:00
// errorKernel is doing all the error handling like what to do if
// an error occurs.
2021-02-05 12:56:42 +00:00
errorKernel * errorKernel
2021-09-15 06:39:34 +00:00
// Ring buffer
ringBuffer * ringBuffer
2021-02-18 11:29:14 +00:00
// metric exporter
metrics * metrics
2021-09-10 08:04:25 +00:00
// Version of package
version string
2022-01-08 03:19:51 +00:00
// tui client
tui * tui
2022-01-19 09:55:03 +00:00
// processInitial is the initial process that all other processes are tied to.
processInitial process
2022-02-04 09:33:31 +00:00
2022-04-21 11:21:36 +00:00
// nodeAuth holds all the signatures, the public keys and other components
// related to authentication on an individual node.
nodeAuth * nodeAuth
2022-04-01 03:46:37 +00:00
// helloRegister is a register of all the nodes that have sent hello messages
// to the central server
helloRegister * helloRegister
// holds the logic for the central auth services
centralAuth * centralAuth
2021-01-27 13:02:57 +00:00
}
2021-01-28 10:17:54 +00:00
// newServer will prepare and return a server type
2022-02-07 03:23:13 +00:00
func NewServer ( configuration * Configuration , version string ) ( * server , error ) {
2021-08-11 10:23:37 +00:00
// Set up the main background context.
2021-07-02 09:26:52 +00:00
ctx , cancel := context . WithCancel ( context . Background ( ) )
2022-02-07 03:23:13 +00:00
metrics := newMetrics ( configuration . PromHostAndPort )
2022-01-19 04:44:20 +00:00
// Start the error kernel that will do all the error handling
// that is not done within a process.
errorKernel := newErrorKernel ( ctx , metrics )
2021-04-19 19:06:37 +00:00
var opt nats . Option
2021-08-16 11:01:12 +00:00
2022-02-07 03:23:13 +00:00
if configuration . RootCAPath != "" {
opt = nats . RootCAs ( configuration . RootCAPath )
2021-04-19 19:06:37 +00:00
}
2022-02-07 03:23:13 +00:00
if configuration . NkeySeedFile != "" {
2021-05-20 10:27:25 +00:00
var err error
2021-08-11 10:23:37 +00:00
2022-02-07 03:23:13 +00:00
opt , err = nats . NkeyOptionFromSeed ( configuration . NkeySeedFile )
2021-05-20 10:27:25 +00:00
if err != nil {
2021-07-02 09:26:52 +00:00
cancel ( )
2021-05-20 10:27:25 +00:00
return nil , fmt . Errorf ( "error: failed to read nkey seed file: %v" , err )
}
}
2021-05-25 08:23:27 +00:00
var conn * nats . Conn
2021-08-16 11:01:12 +00:00
// Connect to the nats server, and retry until succesful.
2021-05-25 08:23:27 +00:00
for {
var err error
// Setting MaxReconnects to -1 which equals unlimited.
2022-02-07 03:23:13 +00:00
conn , err = nats . Connect ( configuration . BrokerAddress ,
2021-12-17 16:28:57 +00:00
opt ,
2022-06-21 08:40:32 +00:00
//nats.FlusherTimeout(time.Second*10),
2021-12-17 16:28:57 +00:00
nats . MaxReconnects ( - 1 ) ,
2022-02-07 03:23:13 +00:00
nats . ReconnectJitter ( time . Duration ( configuration . NatsReconnectJitter ) * time . Millisecond , time . Duration ( configuration . NatsReconnectJitterTLS ) * time . Second ) ,
nats . Timeout ( time . Second * time . Duration ( configuration . NatsConnOptTimeout ) ) ,
2021-12-17 16:28:57 +00:00
)
2021-05-25 08:23:27 +00:00
// If no servers where available, we loop and retry until succesful.
if err != nil {
2022-02-07 03:23:13 +00:00
log . Printf ( "error: could not connect, waiting %v seconds, and retrying: %v\n" , configuration . NatsConnectRetryInterval , err )
time . Sleep ( time . Duration ( time . Second * time . Duration ( configuration . NatsConnectRetryInterval ) ) )
2021-09-01 11:25:02 +00:00
continue
2021-05-25 08:23:27 +00:00
}
2021-02-01 10:13:38 +00:00
2021-05-25 08:23:27 +00:00
break
}
2021-08-11 10:23:37 +00:00
2021-12-16 10:01:01 +00:00
log . Printf ( " * conn.Opts.ReconnectJitterTLS: %v\n" , conn . Opts . ReconnectJitterTLS )
log . Printf ( " * conn.Opts.ReconnectJitter: %v\n" , conn . Opts . ReconnectJitter )
2022-01-07 06:36:19 +00:00
var stewardSocket net . Listener
var err error
2022-12-21 06:17:47 +00:00
// Check if tmp folder for socket exists, if not create it
if _ , err := os . Stat ( configuration . SocketFolder ) ; os . IsNotExist ( err ) {
err := os . MkdirAll ( configuration . SocketFolder , 0700 )
if err != nil {
cancel ( )
return nil , fmt . Errorf ( "error: failed to create socket folder directory %v: %v" , configuration . SocketFolder , err )
}
}
2022-01-07 06:21:21 +00:00
// Open the steward socket file, and start the listener if enabled.
2022-02-07 03:23:13 +00:00
if configuration . EnableSocket {
stewardSocket , err = createSocket ( configuration . SocketFolder , "steward.sock" )
2022-01-07 06:36:19 +00:00
if err != nil {
cancel ( )
return nil , err
}
2021-03-30 08:37:16 +00:00
}
2022-01-08 03:19:51 +00:00
// Create the tui client structure if enabled.
var tuiClient * tui
2022-02-07 03:23:13 +00:00
if configuration . EnableTUI {
tuiClient , err = newTui ( Node ( configuration . NodeName ) )
2022-01-08 03:19:51 +00:00
if err != nil {
cancel ( )
return nil , err
}
}
2022-06-09 06:42:49 +00:00
//var nodeAuth *nodeAuth
//if configuration.EnableSignatureCheck {
2022-04-21 11:21:36 +00:00
nodeAuth := newNodeAuth ( configuration , errorKernel )
2022-02-11 08:04:14 +00:00
// fmt.Printf(" * DEBUG: newServer: signatures contains: %+v\n", signatures)
2022-06-09 06:42:49 +00:00
//}
//var centralAuth *centralAuth
//if configuration.IsCentralAuth {
centralAuth := newCentralAuth ( configuration , errorKernel )
//}
2022-02-04 09:33:31 +00:00
2022-04-01 05:09:55 +00:00
s := server {
ctx : ctx ,
cancel : cancel ,
configuration : configuration ,
nodeName : configuration . NodeName ,
natsConn : conn ,
StewardSocket : stewardSocket ,
toRingBufferCh : make ( chan [ ] subjectAndMessage ) ,
2022-06-20 12:22:42 +00:00
directSAMSCh : make ( chan [ ] subjectAndMessage ) ,
2022-04-01 05:09:55 +00:00
metrics : metrics ,
version : version ,
tui : tuiClient ,
errorKernel : errorKernel ,
2022-04-21 11:21:36 +00:00
nodeAuth : nodeAuth ,
2022-04-01 05:09:55 +00:00
helloRegister : newHelloRegister ( ) ,
2022-06-09 06:42:49 +00:00
centralAuth : centralAuth ,
2021-02-01 12:41:04 +00:00
}
2021-01-29 05:09:48 +00:00
2022-04-01 05:09:55 +00:00
s . processes = newProcesses ( ctx , & s )
2021-03-02 12:46:02 +00:00
// Create the default data folder for where subscribers should
2021-03-25 11:50:58 +00:00
// write it's data, check if data folder exist, and create it if needed.
2022-02-07 03:23:13 +00:00
if _ , err := os . Stat ( configuration . SubscribersDataFolder ) ; os . IsNotExist ( err ) {
if configuration . SubscribersDataFolder == "" {
return nil , fmt . Errorf ( "error: subscribersDataFolder value is empty, you need to provide the config or the flag value at startup %v: %v" , configuration . SubscribersDataFolder , err )
2021-03-02 12:46:02 +00:00
}
2022-02-07 03:23:13 +00:00
err := os . Mkdir ( configuration . SubscribersDataFolder , 0700 )
2021-03-02 12:46:02 +00:00
if err != nil {
2022-02-07 03:23:13 +00:00
return nil , fmt . Errorf ( "error: failed to create data folder directory %v: %v" , configuration . SubscribersDataFolder , err )
2021-03-02 12:46:02 +00:00
}
2022-02-18 08:51:11 +00:00
er := fmt . Errorf ( "info: creating subscribers data folder at %v" , configuration . SubscribersDataFolder )
s . errorKernel . logConsoleOnlyIfDebug ( er , s . configuration )
2021-03-02 12:46:02 +00:00
}
2022-04-01 05:09:55 +00:00
return & s , nil
2021-02-05 06:25:12 +00:00
}
2022-04-01 03:46:37 +00:00
// helloRegister is a register of all the nodes that have sent hello messages.
type helloRegister struct {
}
func newHelloRegister ( ) * helloRegister {
h := helloRegister { }
return & h
}
2022-01-07 06:21:21 +00:00
// create socket will create a socket file, and return the net.Listener to
// communicate with that socket.
func createSocket ( socketFolder string , socketFileName string ) ( net . Listener , error ) {
// Just as an extra check we eventually delete any existing Steward socket files if found.
socketFilepath := filepath . Join ( socketFolder , socketFileName )
if _ , err := os . Stat ( socketFilepath ) ; ! os . IsNotExist ( err ) {
err = os . Remove ( socketFilepath )
if err != nil {
er := fmt . Errorf ( "error: could not delete sock file: %v" , err )
return nil , er
}
}
// Open the socket.
nl , err := net . Listen ( "unix" , socketFilepath )
if err != nil {
er := fmt . Errorf ( "error: failed to open socket: %v" , err )
return nil , er
}
return nl , nil
}
2021-02-24 09:58:02 +00:00
// Start will spawn up all the predefined subscriber processes.
2021-02-10 06:25:44 +00:00
// Spawning of publisher processes is done on the fly by checking
2021-02-24 09:58:02 +00:00
// if there is publisher process for a given message subject, and
2021-08-16 11:01:12 +00:00
// if it does not exist it will spawn one.
2021-02-10 04:11:48 +00:00
func ( s * server ) Start ( ) {
2021-09-23 03:46:25 +00:00
log . Printf ( "Starting steward, version=%+v\n" , s . version )
2021-09-10 08:21:33 +00:00
s . metrics . promVersion . With ( prometheus . Labels { "version" : string ( s . version ) } )
2021-09-10 08:04:25 +00:00
2021-08-04 08:37:24 +00:00
go func ( ) {
2022-04-01 05:09:55 +00:00
err := s . errorKernel . start ( s . toRingBufferCh )
2021-08-04 08:37:24 +00:00
if err != nil {
log . Printf ( "%v\n" , err )
}
} ( )
2021-02-19 10:07:09 +00:00
2021-02-18 11:29:14 +00:00
// Start collecting the metrics
2021-08-03 11:57:29 +00:00
go func ( ) {
2021-08-04 08:37:24 +00:00
err := s . metrics . start ( )
2021-08-03 11:57:29 +00:00
if err != nil {
log . Printf ( "%v\n" , err )
os . Exit ( 1 )
}
} ( )
2021-02-18 11:29:14 +00:00
2021-03-29 11:36:30 +00:00
// Start the checking the input socket for new messages from operator.
2022-01-07 06:36:19 +00:00
if s . configuration . EnableSocket {
go s . readSocket ( )
}
2021-02-26 08:02:53 +00:00
2021-09-10 03:26:16 +00:00
// Check if we should start the tcp listener for new messages from operator.
2021-08-23 14:00:48 +00:00
if s . configuration . TCPListener != "" {
2021-09-09 11:32:04 +00:00
go s . readTCPListener ( )
2021-08-23 14:00:48 +00:00
}
2021-09-10 03:26:16 +00:00
// Check if we should start the http listener for new messages from operator.
if s . configuration . HTTPListener != "" {
go s . readHttpListener ( )
}
2021-08-11 08:11:57 +00:00
// Start up the predefined subscribers.
//
// Since all the logic to handle processes are tied to the process
// struct, we need to create an initial process to start the rest.
2021-08-11 10:23:37 +00:00
//
// NB: The context of the initial process are set in processes.Start.
2021-08-09 07:18:30 +00:00
sub := newSubject ( REQInitial , s . nodeName )
2022-04-01 05:09:55 +00:00
s . processInitial = newProcess ( context . TODO ( ) , s , sub , "" , nil )
2021-08-11 08:11:57 +00:00
// Start all wanted subscriber processes.
2022-01-19 09:55:03 +00:00
s . processes . Start ( s . processInitial )
2021-08-03 11:43:05 +00:00
2021-08-09 07:18:30 +00:00
time . Sleep ( time . Second * 1 )
s . processes . printProcessesMap ( )
2021-02-18 13:27:53 +00:00
2021-08-23 10:47:33 +00:00
// Start exposing the the data folder via HTTP if flag is set.
if s . configuration . ExposeDataFolder != "" {
log . Printf ( "info: Starting expose of data folder via HTTP\n" )
go s . exposeDataFolder ( s . ctx )
}
2022-01-08 03:19:51 +00:00
if s . configuration . EnableTUI {
go func ( ) {
2022-04-01 05:09:55 +00:00
err := s . tui . Start ( s . ctx , s . toRingBufferCh )
2022-01-08 03:19:51 +00:00
if err != nil {
log . Printf ( "%v\n" , err )
os . Exit ( 1 )
}
} ( )
}
2021-08-09 07:18:30 +00:00
// Start the processing of new messages from an input channel.
2021-09-15 06:39:34 +00:00
// NB: We might need to create a sub context for the ringbuffer here
// so we can cancel this context last, and not use the server.
2021-08-25 06:50:24 +00:00
s . routeMessagesToProcess ( "./incomingBuffer.db" )
2021-02-10 06:25:44 +00:00
2022-06-20 12:22:42 +00:00
// Start reading the channel for injecting direct messages that should
// not be sent via the message broker.
s . directSAMSChRead ( )
2022-01-26 08:23:02 +00:00
// Check and enable read the messages specified in the startup folder.
s . readStartupFolder ( )
2021-08-09 07:18:30 +00:00
}
2021-02-05 09:47:07 +00:00
2022-06-20 12:22:42 +00:00
// directSAMSChRead for injecting messages directly in to the local system
// without sending them via the message broker.
func ( s * server ) directSAMSChRead ( ) {
go func ( ) {
2022-09-08 05:37:44 +00:00
for {
select {
case <- s . ctx . Done ( ) :
log . Printf ( "info: stopped the directSAMSCh reader\n\n" )
return
case sams := <- s . directSAMSCh :
2022-11-01 08:54:35 +00:00
// fmt.Printf(" * DEBUG: directSAMSChRead: <- sams = %v\n", sams)
2022-09-08 05:37:44 +00:00
// Range over all the sams, find the process, check if the method exists, and
// handle the message by starting the correct method handler.
for i := range sams {
processName := processNameGet ( sams [ i ] . Subject . name ( ) , processKindSubscriber )
s . processes . active . mu . Lock ( )
p := s . processes . active . procNames [ processName ]
s . processes . active . mu . Unlock ( )
mh , ok := p . methodsAvailable . CheckIfExists ( sams [ i ] . Message . Method )
if ! ok {
er := fmt . Errorf ( "error: subscriberHandler: method type not available: %v" , p . subject . Event )
p . errorKernel . errSend ( p , sams [ i ] . Message , er )
continue
}
p . handler = mh . handler
go executeHandler ( p , sams [ i ] . Message , s . nodeName )
2022-06-20 12:22:42 +00:00
}
}
}
} ( )
}
2021-08-09 07:18:30 +00:00
// Will stop all processes started during startup.
func ( s * server ) Stop ( ) {
// Stop the started pub/sub message processes.
2021-08-11 10:23:37 +00:00
s . processes . Stop ( )
2021-08-09 07:18:30 +00:00
log . Printf ( "info: stopped all subscribers\n" )
2021-07-02 06:38:44 +00:00
2021-08-09 07:18:30 +00:00
// Stop the errorKernel.
s . errorKernel . stop ( )
log . Printf ( "info: stopped the errorKernel\n" )
2021-07-02 11:26:30 +00:00
2021-08-09 07:18:30 +00:00
// Stop the main context.
2021-08-11 10:23:37 +00:00
s . cancel ( )
2021-08-09 07:18:30 +00:00
log . Printf ( "info: stopped the main context\n" )
2021-08-09 12:41:31 +00:00
2022-01-05 18:53:26 +00:00
// Delete the steward socket file when the program exits.
2021-08-09 12:41:31 +00:00
socketFilepath := filepath . Join ( s . configuration . SocketFolder , "steward.sock" )
if _ , err := os . Stat ( socketFilepath ) ; ! os . IsNotExist ( err ) {
err = os . Remove ( socketFilepath )
if err != nil {
er := fmt . Errorf ( "error: could not delete sock file: %v" , err )
log . Printf ( "%v\n" , er )
}
}
2021-02-05 09:47:07 +00:00
}
2022-06-17 22:15:18 +00:00
// samDBValueAndDelivered Contains the sam value as it is used in the
// state DB, and also a delivered function to be called when this message
// is picked up, so we can control if messages gets stale at some point.
2021-07-02 17:09:42 +00:00
type samDBValueAndDelivered struct {
2021-07-02 16:32:01 +00:00
samDBValue samDBValue
2021-07-05 05:43:33 +00:00
delivered func ( )
2021-07-02 16:32:01 +00:00
}
2021-08-25 06:50:24 +00:00
// routeMessagesToProcess takes a database name it's input argument.
// The database will be used as the persistent k/v store for the work
// queue which is implemented as a ring buffer.
2022-01-20 07:17:37 +00:00
// The ringBufferInCh are where we get new messages to publish.
2021-03-09 06:43:55 +00:00
// Incomming messages will be routed to the correct subject process, where
// the handling of each nats subject is handled within it's own separate
// worker process.
// It will also handle the process of spawning more worker processes
// for publisher subjects if it does not exist.
2021-08-25 06:50:24 +00:00
func ( s * server ) routeMessagesToProcess ( dbFileName string ) {
2021-03-09 06:43:55 +00:00
// Prepare and start a new ring buffer
2021-12-21 06:21:12 +00:00
var bufferSize int = s . configuration . RingBufferSize
2021-09-14 14:23:01 +00:00
const samValueBucket string = "samValueBucket"
const indexValueBucket string = "indexValueBucket"
2022-04-01 05:09:55 +00:00
s . ringBuffer = newringBuffer ( s . ctx , s . metrics , s . configuration , bufferSize , dbFileName , Node ( s . nodeName ) , s . toRingBufferCh , samValueBucket , indexValueBucket , s . errorKernel , s . processInitial )
2021-08-26 05:02:36 +00:00
2021-08-25 06:31:48 +00:00
ringBufferInCh := make ( chan subjectAndMessage )
2021-07-02 17:09:42 +00:00
ringBufferOutCh := make ( chan samDBValueAndDelivered )
2021-03-09 06:43:55 +00:00
// start the ringbuffer.
2021-11-18 05:50:25 +00:00
s . ringBuffer . start ( s . ctx , ringBufferInCh , ringBufferOutCh )
2021-03-09 06:43:55 +00:00
// Start reading new fresh messages received on the incomming message
// pipe/file requested, and fill them into the buffer.
2022-01-20 06:50:58 +00:00
// Since the new messages comming into the system is a []subjectAndMessage
// we loop here, unfold the slice, and put single subjectAndMessages's on
// the channel to the ringbuffer.
2021-03-09 06:43:55 +00:00
go func ( ) {
2022-04-01 05:09:55 +00:00
for sams := range s . toRingBufferCh {
2021-08-25 08:16:55 +00:00
for _ , sam := range sams {
2021-08-25 06:31:48 +00:00
ringBufferInCh <- sam
2021-03-09 06:43:55 +00:00
}
}
2021-08-25 06:31:48 +00:00
close ( ringBufferInCh )
2021-03-09 06:43:55 +00:00
} ( )
// Process the messages that are in the ring buffer. Check and
// send if there are a specific subject for it, and if no subject
// exist throw an error.
2022-01-27 06:19:04 +00:00
var event Event
eventAvailable := event . CheckEventAvailable ( )
2021-03-09 06:43:55 +00:00
var method Method
methodsAvailable := method . GetMethodsAvailable ( )
go func ( ) {
2021-11-09 12:18:58 +00:00
for samDBVal := range ringBufferOutCh {
2022-12-29 06:53:25 +00:00
go func ( samDBVal samDBValueAndDelivered ) {
// Signal back to the ringbuffer that message have been picked up.
samDBVal . delivered ( )
sam := samDBVal . samDBValue . Data
// Check if the format of the message is correct.
if _ , ok := methodsAvailable . CheckIfExists ( sam . Message . Method ) ; ! ok {
er := fmt . Errorf ( "error: routeMessagesToProcess: the method do not exist, message dropped: %v" , sam . Message . Method )
s . errorKernel . errSend ( s . processInitial , sam . Message , er )
return
}
if ! eventAvailable . CheckIfExists ( sam . Subject . Event , sam . Subject ) {
er := fmt . Errorf ( "error: routeMessagesToProcess: the event type do not exist, message dropped: %v" , sam . Message . Method )
s . errorKernel . errSend ( s . processInitial , sam . Message , er )
2021-03-12 11:08:11 +00:00
2022-12-29 06:53:25 +00:00
return
}
2021-03-09 06:43:55 +00:00
2022-12-29 06:53:25 +00:00
for {
// Looping here so we are able to redo the sending
// of the last message if a process for the specified subject
// is not present. The process will then be created, and
// the code will loop back here.
m := sam . Message
2021-11-09 13:01:42 +00:00
2022-12-29 06:53:25 +00:00
subjName := sam . Subject . name ( )
pn := processNameGet ( subjName , processKindPublisher )
2021-09-21 21:29:42 +00:00
2022-12-29 06:53:25 +00:00
// Check if there is a map of type map[int]process registered
// for the processName, and if it exists then return it.
s . processes . active . mu . Lock ( )
proc , ok := s . processes . active . procNames [ pn ]
s . processes . active . mu . Unlock ( )
// If found a map above, range it, and are there already a process
// for that subject, put the message on that processes incomming
// message channel.
if ok {
// We have found the process to route the message to, deliver it.
proc . subject . messageCh <- m
break
} else {
// If a publisher process do not exist for the given subject, create it.
// log.Printf("info: processNewMessages: did not find that specific subject, starting new process for subject: %v\n", subjName)
sub := newSubject ( sam . Subject . Method , sam . Subject . ToNode )
var proc process
switch {
case m . IsSubPublishedMsg :
proc = newSubProcess ( s . ctx , s , sub , processKindPublisher , nil )
default :
proc = newProcess ( s . ctx , s , sub , processKindPublisher , nil )
}
proc . spawnWorker ( )
er := fmt . Errorf ( "info: processNewMessages: new process started, subject: %v, processID: %v" , subjName , proc . processID )
s . errorKernel . logConsoleOnlyIfDebug ( er , s . configuration )
// Now when the process is spawned we continue,
// and send the message to that new process.
continue
}
2021-09-21 21:29:42 +00:00
}
2022-12-29 06:53:25 +00:00
} ( samDBVal )
2021-03-09 06:43:55 +00:00
}
} ( )
}
2021-08-23 10:47:33 +00:00
func ( s * server ) exposeDataFolder ( ctx context . Context ) {
2021-08-23 15:05:56 +00:00
fileHandler := func ( w http . ResponseWriter , r * http . Request ) {
// w.Header().Set("Content-Type", "text/html")
http . FileServer ( http . Dir ( s . configuration . SubscribersDataFolder ) ) . ServeHTTP ( w , r )
}
2021-08-23 10:47:33 +00:00
//create a file server, and serve the files found in ./
2021-08-23 15:05:56 +00:00
//fd := http.FileServer(http.Dir(s.configuration.SubscribersDataFolder))
http . HandleFunc ( "/" , fileHandler )
2021-08-23 10:47:33 +00:00
// we create a net.Listen type to use later with the http.Serve function.
nl , err := net . Listen ( "tcp" , s . configuration . ExposeDataFolder )
if err != nil {
log . Println ( "error: starting net.Listen: " , err )
}
// start the web server with http.Serve instead of the usual http.ListenAndServe
err = http . Serve ( nl , nil )
if err != nil {
log . Printf ( "Error: failed to start web server: %v\n" , err )
}
os . Exit ( 1 )
}