Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aebcb615a5 | ||
|
|
7901e2ebee |
No files matched your search
+1
-1
@@ -42,6 +42,6 @@ channel=CHANNEL
|
||||
checksum=[the SHA512 checksum of the message]
|
||||
|
||||
Constrains to the environment:
|
||||
The three field names 'command', 'channel' and 'checksum' are reserved, thus, you cannot use these names.
|
||||
The web server cannot reorder the fields of the request or response. The order of fields at the data object (message) must correspond with the order of fields inside the HTTP message. Therefore, a reorder is not possible at the moment. Also, the three field names 'command', 'channel' and 'checksum' are reserved, thus, you cannot use these names.
|
||||
*/
|
||||
package ICCC
|
||||
@@ -35,7 +35,6 @@ func ICCCGetHostsReceiver(data map[string][]string) (result map[string][]string)
|
||||
answerMessage := SystemMessages.ICCCGetHostsAnswer{}
|
||||
answerMessage.Hostnames = make([]string, countHosts, countHosts)
|
||||
answerMessage.IPAddressesPorts = make([]string, countHosts, countHosts)
|
||||
answerMessage.Kinds = make([]byte, countHosts, countHosts)
|
||||
|
||||
// Loop over all hosts which are currently available at the cache:
|
||||
n := 0
|
||||
@@ -43,7 +42,6 @@ func ICCCGetHostsReceiver(data map[string][]string) (result map[string][]string)
|
||||
host := entry.Value.(Scheme.Host)
|
||||
answerMessage.Hostnames[n] = host.Hostname
|
||||
answerMessage.IPAddressesPorts[n] = host.IPAddressPort
|
||||
answerMessage.Kinds[n] = host.Kind
|
||||
n++
|
||||
}
|
||||
|
||||
|
||||
@@ -36,7 +36,6 @@ func ICCCGetListenersReceiver(data map[string][]string) (result map[string][]str
|
||||
answerMessage.Channels = make([]string, countListeners, countListeners)
|
||||
answerMessage.Commands = make([]string, countListeners, countListeners)
|
||||
answerMessage.IPAddressesPorts = make([]string, countListeners, countListeners)
|
||||
answerMessage.Kinds = make([]byte, countListeners, countListeners)
|
||||
|
||||
// Loop over all hosts which are currently available at the cache:
|
||||
n := 0
|
||||
@@ -45,7 +44,6 @@ func ICCCGetListenersReceiver(data map[string][]string) (result map[string][]str
|
||||
answerMessage.Channels[n] = listener.Channel
|
||||
answerMessage.Commands[n] = listener.Command
|
||||
answerMessage.IPAddressesPorts[n] = listener.IPAddressPort
|
||||
answerMessage.Kinds[n] = listener.Kind
|
||||
n++
|
||||
}
|
||||
|
||||
|
||||
@@ -28,7 +28,7 @@ func ICCCGetVersionReceiver(data map[string][]string) (result map[string][]strin
|
||||
|
||||
// Prepare the answer:
|
||||
answer := SystemMessages.ICCCGetVersionAnswer{}
|
||||
answer.Kind = KindOCEAN
|
||||
answer.Kind = `Ocean`
|
||||
answer.Name = Tools.ThisHostname()
|
||||
answer.Version = Version.GetVersion()
|
||||
|
||||
|
||||
@@ -39,7 +39,6 @@ func ICCCListenerUpdateReceiver(data map[string][]string) (result map[string][]s
|
||||
updatedObject.Command = messageData.Command
|
||||
updatedObject.IPAddressPort = messageData.IPAddressPort
|
||||
updatedObject.IsActive = messageData.IsActiveNew
|
||||
updatedObject.Kind = messageData.Kind
|
||||
|
||||
// Update the entry:
|
||||
if errUpdate := collectionListener.Update(selectionUpdate, updatedObject); errUpdate != nil {
|
||||
|
||||
@@ -31,7 +31,7 @@ func ICCCRegisterHostReceiver(data map[string][]string) (result map[string][]str
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: Register another host.`, fmt.Sprintf("hostname=%s", messageData.Hostname), fmt.Sprintf("ipAddressPort=%s", messageData.IPAddressPort))
|
||||
|
||||
// Execute the command:
|
||||
registerHost2Database(messageData.Hostname, messageData.IPAddressPort, messageData.Kind)
|
||||
registerHost2Database(messageData.Hostname, messageData.IPAddressPort)
|
||||
|
||||
// Update the caches:
|
||||
InitCacheNow()
|
||||
|
||||
@@ -31,7 +31,7 @@ func ICCCRegisterListenerReceiver(data map[string][]string) (result map[string][
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: Should register another listener.`, `channel=`+messageData.Channel, `command=`+messageData.Command, `IPAddressPort=`+messageData.IPAddressPort, fmt.Sprintf(`isActive=%v`, messageData.IsActive))
|
||||
|
||||
// Execute the command:
|
||||
registerListener2Database(messageData.Channel, messageData.Command, messageData.IPAddressPort, messageData.IsActive, messageData.Kind)
|
||||
registerListener2Database(messageData.Channel, messageData.Command, messageData.IPAddressPort, messageData.IsActive)
|
||||
|
||||
// Update the caches:
|
||||
InitCacheNow()
|
||||
|
||||
@@ -28,7 +28,6 @@ func initDB() {
|
||||
//
|
||||
// Take care about the indexes for ICCCListener:
|
||||
//
|
||||
collectionListener.EnsureIndexKey(`Kind`)
|
||||
collectionListener.EnsureIndexKey(`Command`)
|
||||
collectionListener.EnsureIndexKey(`Command`, `IsActive`)
|
||||
|
||||
@@ -53,7 +52,6 @@ func initDB() {
|
||||
//
|
||||
// Index for hosts:
|
||||
//
|
||||
collectionHosts.EnsureIndexKey(`Kind`)
|
||||
collectionHosts.EnsureIndexKey(`Hostname`)
|
||||
collectionHosts.EnsureIndexKey(`IPAddressPort`)
|
||||
|
||||
|
||||
@@ -8,13 +8,12 @@ import (
|
||||
)
|
||||
|
||||
// Function to register a server to the ICCC.
|
||||
func registerHost2Database(hostname, ipAddressPort string, kind byte) {
|
||||
func registerHost2Database(hostname, ipAddressPort string) {
|
||||
|
||||
// Create the host entry:
|
||||
host := Scheme.Host{}
|
||||
host.Hostname = hostname
|
||||
host.IPAddressPort = ipAddressPort
|
||||
host.Kind = kind
|
||||
|
||||
// The query to find already existing entries:
|
||||
selection := bson.D{{`Hostname`, host.Hostname}, {`IPAddressPort`, host.IPAddressPort}}
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
// The internal function to register an listener to ICCC.
|
||||
func registerListener2Database(channel, command, ipAddressPort string, isActive bool, kind byte) {
|
||||
func registerListener2Database(channel, command, ipAddressPort string, isActive bool) {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `Register this ICCC command in to the database.`, `channel=`+channel, `command=`+command, `IPAddressPort=`+ipAddressPort, fmt.Sprintf("isActive=%v", isActive))
|
||||
|
||||
entry := Scheme.Listener{}
|
||||
@@ -17,7 +17,6 @@ func registerListener2Database(channel, command, ipAddressPort string, isActive
|
||||
entry.Command = command
|
||||
entry.IsActive = isActive
|
||||
entry.IPAddressPort = ipAddressPort
|
||||
entry.Kind = kind
|
||||
|
||||
//
|
||||
// Case: Exists?
|
||||
|
||||
@@ -8,5 +8,5 @@ func registerLocalListener2Database(channel, command string) {
|
||||
there would be no server which can execute the ICCC command.
|
||||
Therefore, every Ocean server registers the own listeners directly.
|
||||
*/
|
||||
registerListener2Database(channel, command, correctAddressWithPort, true, KindOCEAN)
|
||||
registerListener2Database(channel, command, correctAddressWithPort, true)
|
||||
}
|
||||
@@ -12,5 +12,5 @@ func registerThisHost2Database() {
|
||||
there would be no server which can execute the ICCC command.
|
||||
Therefore, every Ocean server registers the own host directly.
|
||||
*/
|
||||
registerHost2Database(Tools.ThisHostname(), correctAddressWithPort, KindOCEAN)
|
||||
registerHost2Database(Tools.ThisHostname(), correctAddressWithPort)
|
||||
}
|
||||
@@ -4,5 +4,4 @@ package Scheme
|
||||
type Host struct {
|
||||
Hostname string `bson:"Hostname"`
|
||||
IPAddressPort string `bson:"IPAddressPort"`
|
||||
Kind byte `bson:"Kind"`
|
||||
}
|
||||
@@ -6,5 +6,4 @@ type Listener struct {
|
||||
Command string `bson:"Command"`
|
||||
IsActive bool `bson:"IsActive"`
|
||||
IPAddressPort string `bson:"IPAddressPort"`
|
||||
Kind byte `bson:"Kind"`
|
||||
}
|
||||
+2
-3
@@ -6,7 +6,6 @@ import (
|
||||
"github.com/SommerEngineering/Ocean/Tools"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Sign a message to secure it.
|
||||
@@ -33,7 +32,7 @@ func signMessage(data map[string][]string) (result url.Values) {
|
||||
// Now, loop over all the data:
|
||||
for _, key := range keys {
|
||||
// Get the value:
|
||||
value := data[key][0]
|
||||
value := data[key]
|
||||
|
||||
// Hash each key and value:
|
||||
fmt.Fprintf(hash, "key=%s :: value=%s\n", key, value)
|
||||
@@ -43,6 +42,6 @@ func signMessage(data map[string][]string) (result url.Values) {
|
||||
result = url.Values(data)
|
||||
|
||||
// Append the sign:
|
||||
result.Add(`checksum`, strings.ToUpper(fmt.Sprintf("%x", hash.Sum(nil))))
|
||||
result.Add(`checksum`, fmt.Sprintf("%x", hash.Sum(nil)))
|
||||
return
|
||||
}
|
||||
@@ -8,5 +8,4 @@ type ICCCGetHosts struct {
|
||||
type ICCCGetHostsAnswer struct {
|
||||
Hostnames []string
|
||||
IPAddressesPorts []string
|
||||
Kinds []byte
|
||||
}
|
||||
@@ -9,5 +9,4 @@ type ICCCGetListenersAnswer struct {
|
||||
Channels []string
|
||||
Commands []string
|
||||
IPAddressesPorts []string
|
||||
Kinds []byte
|
||||
}
|
||||
@@ -6,7 +6,7 @@ type ICCCGetVersion struct {
|
||||
|
||||
// Answer to the version request
|
||||
type ICCCGetVersionAnswer struct {
|
||||
Kind byte // Ocean || Component
|
||||
Kind string // Ocean || Component
|
||||
Name string // Ocean: Hostname; Components: Name
|
||||
Version string // The current version
|
||||
}
|
||||
@@ -6,5 +6,4 @@ type ICCCListenerUpdate struct {
|
||||
Command string
|
||||
IPAddressPort string
|
||||
IsActiveNew bool
|
||||
Kind byte
|
||||
}
|
||||
@@ -1,15 +0,0 @@
|
||||
package SystemMessages
|
||||
|
||||
// The message to send logging events over ICCC.
|
||||
type ICCCNewLogEvent struct {
|
||||
Project string
|
||||
UnixTimestampUTC int64
|
||||
Sender string
|
||||
Category string
|
||||
Level string
|
||||
Severity string
|
||||
Impact string
|
||||
MessageName string
|
||||
MessageDescription string
|
||||
Parameters []string
|
||||
}
|
||||
@@ -4,5 +4,4 @@ package SystemMessages
|
||||
type ICCCRegisterHost struct {
|
||||
Hostname string // The hostname for the end-point
|
||||
IPAddressPort string // The IP address and port for the end-point
|
||||
Kind byte // Ocean || Component
|
||||
}
|
||||
@@ -6,5 +6,4 @@ type ICCCRegisterListener struct {
|
||||
Command string // The provided command
|
||||
IsActive bool // Is the command active?
|
||||
IPAddressPort string // The IP address and port for the end-point
|
||||
Kind byte // Ocean || Component
|
||||
}
|
||||
@@ -15,10 +15,6 @@ const (
|
||||
ChannelSTARTUP string = `System::Startup` // A channel for system startup messages.
|
||||
ChannelICCC string = `System::ICCC` // A common ICCC channel.
|
||||
ChannelPING string = `System::Ping` // A channel for pings.
|
||||
ChannelLOGGING string = `System::Logging` // A channel for send log events to the logging system
|
||||
KindOCEAN byte = 0x0
|
||||
KindComponent byte = 0x1
|
||||
KindALL byte = 0xFF
|
||||
)
|
||||
|
||||
var (
|
||||
|
||||
@@ -7,7 +7,7 @@ import (
|
||||
)
|
||||
|
||||
// Function to broadcast a message to all listeners.
|
||||
func WriteMessage2All(channel, command string, kind byte, message interface{}, answerPrototype interface{}) (results []interface{}) {
|
||||
func WriteMessage2All(channel, command string, message interface{}, answerPrototype interface{}) (results []interface{}) {
|
||||
cacheListenerDatabaseLock.RLock()
|
||||
defer cacheListenerDatabaseLock.RUnlock()
|
||||
|
||||
@@ -21,15 +21,9 @@ func WriteMessage2All(channel, command string, kind byte, message interface{}, a
|
||||
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
|
||||
listener := entry.Value.(Scheme.Listener)
|
||||
|
||||
// If the channel, command and kind matches, deliver the message:
|
||||
if kind == KindALL {
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
matchingListener = append(matchingListener, listener)
|
||||
}
|
||||
} else {
|
||||
if listener.Channel == channel && listener.Command == command && listener.Kind == kind {
|
||||
matchingListener = append(matchingListener, listener)
|
||||
}
|
||||
// If the channel and the command matches, deliver the message:
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
matchingListener = append(matchingListener, listener)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
)
|
||||
|
||||
// Function to write a message to any listener.
|
||||
func WriteMessage2Any(channel, command string, kind byte, message interface{}, answerPrototype interface{}) (result interface{}) {
|
||||
func WriteMessage2Any(channel, command string, message interface{}, answerPrototype interface{}) (result interface{}) {
|
||||
cacheListenerDatabaseLock.RLock()
|
||||
defer cacheListenerDatabaseLock.RUnlock()
|
||||
|
||||
@@ -22,17 +22,10 @@ func WriteMessage2Any(channel, command string, kind byte, message interface{}, a
|
||||
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
|
||||
listener := entry.Value.(Scheme.Listener)
|
||||
|
||||
// If the channel, command and kind matches, store the listener:
|
||||
if kind == KindALL {
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
entries = entries[:len(entries)+1]
|
||||
entries[counter] = listener
|
||||
}
|
||||
} else {
|
||||
if listener.Channel == channel && listener.Command == command && listener.Kind == kind {
|
||||
entries = entries[:len(entries)+1]
|
||||
entries[counter] = listener
|
||||
}
|
||||
// If the channel and the command matches, store the listener:
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
entries = entries[:len(entries)+1]
|
||||
entries[counter] = listener
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +0,0 @@
|
||||
package Log
|
||||
|
||||
// This function provides the project name for e.g. logging events
|
||||
// coming trough the ICCC.
|
||||
func GetProjectName() string {
|
||||
return projectName
|
||||
}
|
||||
+1
-7
@@ -121,16 +121,10 @@ func clearEntry(entry Meta.Entry) (result Meta.Entry) {
|
||||
// Remove white spaces from the parameters.
|
||||
func clearParameters(oldParameters []string) (result []string) {
|
||||
for n := 0; n < len(oldParameters); n++ {
|
||||
// Remove white spaces:
|
||||
oldParameters[n] = removeWhitespaces(oldParameters[n])
|
||||
|
||||
// Is content left after removing?
|
||||
if oldParameters[n] != `` {
|
||||
// Yes, some content is still there. Thus, commit this entry:
|
||||
result = append(result, oldParameters[n])
|
||||
}
|
||||
}
|
||||
|
||||
result = oldParameters
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
package ICCCLog
|
||||
|
||||
import (
|
||||
"github.com/SommerEngineering/Ocean/ICCC"
|
||||
"github.com/SommerEngineering/Ocean/ICCC/SystemMessages"
|
||||
"github.com/SommerEngineering/Ocean/Log"
|
||||
LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The receiver function for the ICCC new log event message.
|
||||
func ICCCNewLogEventReceiver(data map[string][]string) (result map[string][]string) {
|
||||
|
||||
// Recover from errors:
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityUnknown, LM.ImpactUnknown, LM.MessageNamePARSE, "Was not able to execute the ICCC new log event message.")
|
||||
result = make(map[string][]string, 0)
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
// Converts the HTTP form data into an object:
|
||||
_, _, obj := ICCC.Data2Message(SystemMessages.ICCCNewLogEvent{}, data)
|
||||
|
||||
// Was it possible to convert the data?
|
||||
if obj != nil {
|
||||
// Cast the object to the right type:
|
||||
messageData := obj.(SystemMessages.ICCCNewLogEvent)
|
||||
|
||||
// Provide a log entry:
|
||||
logEntry := LM.Entry{}
|
||||
logEntry.Project = Log.GetProjectName()
|
||||
logEntry.Sender = LM.Sender(messageData.Sender)
|
||||
logEntry.Category = LM.ParseCategory(messageData.Category)
|
||||
logEntry.Level = LM.ParseLevel(messageData.Level)
|
||||
logEntry.Severity = LM.ParseSeverity(messageData.Severity)
|
||||
logEntry.Impact = LM.ParseImpact(messageData.Impact)
|
||||
logEntry.MessageName = LM.MessageName(messageData.MessageName)
|
||||
logEntry.MessageDescription = messageData.MessageDescription
|
||||
logEntry.Parameters = messageData.Parameters
|
||||
logEntry.Time = time.Unix(messageData.UnixTimestampUTC, 0).UTC()
|
||||
|
||||
// Deliver the log event:
|
||||
Log.TakeEntry(logEntry)
|
||||
|
||||
// An answer is necessary:
|
||||
return ICCC.Message2Data("", "", SystemMessages.AnswerACK)
|
||||
} else {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: Was not able to create the message.`)
|
||||
}
|
||||
|
||||
// In any other error case:
|
||||
result = make(map[string][]string, 0)
|
||||
return
|
||||
}
|
||||
@@ -1,9 +0,0 @@
|
||||
package ICCCLog
|
||||
|
||||
import (
|
||||
"github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
)
|
||||
|
||||
var (
|
||||
senderName Meta.Sender = `System::Log::ICCCLog`
|
||||
)
|
||||
@@ -6,7 +6,6 @@ import (
|
||||
"github.com/SommerEngineering/Ocean/CustomerDB"
|
||||
"github.com/SommerEngineering/Ocean/ICCC"
|
||||
"github.com/SommerEngineering/Ocean/Log"
|
||||
"github.com/SommerEngineering/Ocean/Log/ICCCLog"
|
||||
LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
"github.com/SommerEngineering/Ocean/NumGen"
|
||||
"github.com/SommerEngineering/Ocean/Shutdown"
|
||||
@@ -110,7 +109,6 @@ func initSystem() {
|
||||
ICCC.Registrar(ICCC.ChannelICCC, `ICCC::GetHosts`, ICCC.ICCCGetHostsReceiver)
|
||||
ICCC.Registrar(ICCC.ChannelICCC, `ICCC::GetListeners`, ICCC.ICCCGetListenersReceiver)
|
||||
ICCC.Registrar(ICCC.ChannelNUMGEN, `NumGen::Next`, NumGen.ICCCNextNumberReceiver)
|
||||
ICCC.Registrar(ICCC.ChannelLOGGING, `Logging::NewLogEvent`, ICCCLog.ICCCNewLogEventReceiver)
|
||||
|
||||
// Start the ICCC Listener Cache:
|
||||
ICCC.InitCacheNow() // Blocking, until the job is done
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
package Version
|
||||
|
||||
var (
|
||||
oceansVersion string = `2.0.4` // Ocean's current version
|
||||
oceansVersion string = `2.0.0` // Ocean's current version
|
||||
)
|
||||
+1
-1
@@ -30,7 +30,7 @@ func Start() {
|
||||
}
|
||||
|
||||
// Notify the whole cluster, that this server is now up and ready:
|
||||
answers := ICCC.WriteMessage2All(ICCC.ChannelSTARTUP, `System::OceanStart`, ICCC.KindALL, data, SystemMessages.ICCCDefaultAnswer{})
|
||||
answers := ICCC.WriteMessage2All(ICCC.ChannelSTARTUP, `System::OceanStart`, data, SystemMessages.ICCCDefaultAnswer{})
|
||||
for n, obj := range answers {
|
||||
if obj != nil {
|
||||
answer := obj.(SystemMessages.ICCCDefaultAnswer)
|
||||
|
||||
Reference in new issue
Block a user