Author SHA1 Message Date
Thorsten Sommer 6ab66faa9e Fixed issue with ICCCProcessor 2015-07-12 12:17:16 +02:00
32 changed files with 20 additions and 67 deletions

No files matched your search

+1 -1
View File
@@ -42,6 +42,6 @@ channel=CHANNEL
checksum=[the SHA512 checksum of the message] checksum=[the SHA512 checksum of the message]
Constrains to the environment: 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 package ICCC
-2
View File
@@ -35,7 +35,6 @@ func ICCCGetHostsReceiver(data map[string][]string) (result map[string][]string)
answerMessage := SystemMessages.ICCCGetHostsAnswer{} answerMessage := SystemMessages.ICCCGetHostsAnswer{}
answerMessage.Hostnames = make([]string, countHosts, countHosts) answerMessage.Hostnames = make([]string, countHosts, countHosts)
answerMessage.IPAddressesPorts = 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: // Loop over all hosts which are currently available at the cache:
n := 0 n := 0
@@ -43,7 +42,6 @@ func ICCCGetHostsReceiver(data map[string][]string) (result map[string][]string)
host := entry.Value.(Scheme.Host) host := entry.Value.(Scheme.Host)
answerMessage.Hostnames[n] = host.Hostname answerMessage.Hostnames[n] = host.Hostname
answerMessage.IPAddressesPorts[n] = host.IPAddressPort answerMessage.IPAddressesPorts[n] = host.IPAddressPort
answerMessage.Kinds[n] = host.Kind
n++ n++
} }
-2
View File
@@ -36,7 +36,6 @@ func ICCCGetListenersReceiver(data map[string][]string) (result map[string][]str
answerMessage.Channels = make([]string, countListeners, countListeners) answerMessage.Channels = make([]string, countListeners, countListeners)
answerMessage.Commands = make([]string, countListeners, countListeners) answerMessage.Commands = make([]string, countListeners, countListeners)
answerMessage.IPAddressesPorts = 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: // Loop over all hosts which are currently available at the cache:
n := 0 n := 0
@@ -45,7 +44,6 @@ func ICCCGetListenersReceiver(data map[string][]string) (result map[string][]str
answerMessage.Channels[n] = listener.Channel answerMessage.Channels[n] = listener.Channel
answerMessage.Commands[n] = listener.Command answerMessage.Commands[n] = listener.Command
answerMessage.IPAddressesPorts[n] = listener.IPAddressPort answerMessage.IPAddressesPorts[n] = listener.IPAddressPort
answerMessage.Kinds[n] = listener.Kind
n++ n++
} }
+1 -1
View File
@@ -28,7 +28,7 @@ func ICCCGetVersionReceiver(data map[string][]string) (result map[string][]strin
// Prepare the answer: // Prepare the answer:
answer := SystemMessages.ICCCGetVersionAnswer{} answer := SystemMessages.ICCCGetVersionAnswer{}
answer.Kind = KindOCEAN answer.Kind = `Ocean`
answer.Name = Tools.ThisHostname() answer.Name = Tools.ThisHostname()
answer.Version = Version.GetVersion() answer.Version = Version.GetVersion()
-1
View File
@@ -39,7 +39,6 @@ func ICCCListenerUpdateReceiver(data map[string][]string) (result map[string][]s
updatedObject.Command = messageData.Command updatedObject.Command = messageData.Command
updatedObject.IPAddressPort = messageData.IPAddressPort updatedObject.IPAddressPort = messageData.IPAddressPort
updatedObject.IsActive = messageData.IsActiveNew updatedObject.IsActive = messageData.IsActiveNew
updatedObject.Kind = messageData.Kind
// Update the entry: // Update the entry:
if errUpdate := collectionListener.Update(selectionUpdate, updatedObject); errUpdate != nil { if errUpdate := collectionListener.Update(selectionUpdate, updatedObject); errUpdate != nil {
+1 -1
View File
@@ -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)) 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: // Execute the command:
registerHost2Database(messageData.Hostname, messageData.IPAddressPort, messageData.Kind) registerHost2Database(messageData.Hostname, messageData.IPAddressPort)
// Update the caches: // Update the caches:
InitCacheNow() InitCacheNow()
+1 -1
View File
@@ -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)) 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: // 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: // Update the caches:
InitCacheNow() InitCacheNow()
-2
View File
@@ -28,7 +28,6 @@ func initDB() {
// //
// Take care about the indexes for ICCCListener: // Take care about the indexes for ICCCListener:
// //
collectionListener.EnsureIndexKey(`Kind`)
collectionListener.EnsureIndexKey(`Command`) collectionListener.EnsureIndexKey(`Command`)
collectionListener.EnsureIndexKey(`Command`, `IsActive`) collectionListener.EnsureIndexKey(`Command`, `IsActive`)
@@ -53,7 +52,6 @@ func initDB() {
// //
// Index for hosts: // Index for hosts:
// //
collectionHosts.EnsureIndexKey(`Kind`)
collectionHosts.EnsureIndexKey(`Hostname`) collectionHosts.EnsureIndexKey(`Hostname`)
collectionHosts.EnsureIndexKey(`IPAddressPort`) collectionHosts.EnsureIndexKey(`IPAddressPort`)
+1 -2
View File
@@ -8,13 +8,12 @@ import (
) )
// Function to register a server to the ICCC. // Function to register a server to the ICCC.
func registerHost2Database(hostname, ipAddressPort string, kind byte) { func registerHost2Database(hostname, ipAddressPort string) {
// Create the host entry: // Create the host entry:
host := Scheme.Host{} host := Scheme.Host{}
host.Hostname = hostname host.Hostname = hostname
host.IPAddressPort = ipAddressPort host.IPAddressPort = ipAddressPort
host.Kind = kind
// The query to find already existing entries: // The query to find already existing entries:
selection := bson.D{{`Hostname`, host.Hostname}, {`IPAddressPort`, host.IPAddressPort}} selection := bson.D{{`Hostname`, host.Hostname}, {`IPAddressPort`, host.IPAddressPort}}
+1 -2
View File
@@ -9,7 +9,7 @@ import (
) )
// The internal function to register an listener to ICCC. // 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)) 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{} entry := Scheme.Listener{}
@@ -17,7 +17,6 @@ func registerListener2Database(channel, command, ipAddressPort string, isActive
entry.Command = command entry.Command = command
entry.IsActive = isActive entry.IsActive = isActive
entry.IPAddressPort = ipAddressPort entry.IPAddressPort = ipAddressPort
entry.Kind = kind
// //
// Case: Exists? // Case: Exists?
+1 -1
View File
@@ -8,5 +8,5 @@ func registerLocalListener2Database(channel, command string) {
there would be no server which can execute the ICCC command. there would be no server which can execute the ICCC command.
Therefore, every Ocean server registers the own listeners directly. Therefore, every Ocean server registers the own listeners directly.
*/ */
registerListener2Database(channel, command, correctAddressWithPort, true, KindOCEAN) registerListener2Database(channel, command, correctAddressWithPort, true)
} }
+1 -1
View File
@@ -12,5 +12,5 @@ func registerThisHost2Database() {
there would be no server which can execute the ICCC command. there would be no server which can execute the ICCC command.
Therefore, every Ocean server registers the own host directly. Therefore, every Ocean server registers the own host directly.
*/ */
registerHost2Database(Tools.ThisHostname(), correctAddressWithPort, KindOCEAN) registerHost2Database(Tools.ThisHostname(), correctAddressWithPort)
} }
-1
View File
@@ -4,5 +4,4 @@ package Scheme
type Host struct { type Host struct {
Hostname string `bson:"Hostname"` Hostname string `bson:"Hostname"`
IPAddressPort string `bson:"IPAddressPort"` IPAddressPort string `bson:"IPAddressPort"`
Kind byte `bson:"Kind"`
} }
-1
View File
@@ -6,5 +6,4 @@ type Listener struct {
Command string `bson:"Command"` Command string `bson:"Command"`
IsActive bool `bson:"IsActive"` IsActive bool `bson:"IsActive"`
IPAddressPort string `bson:"IPAddressPort"` IPAddressPort string `bson:"IPAddressPort"`
Kind byte `bson:"Kind"`
} }
+2 -3
View File
@@ -6,7 +6,6 @@ import (
"github.com/SommerEngineering/Ocean/Tools" "github.com/SommerEngineering/Ocean/Tools"
"net/url" "net/url"
"sort" "sort"
"strings"
) )
// Sign a message to secure it. // 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: // Now, loop over all the data:
for _, key := range keys { for _, key := range keys {
// Get the value: // Get the value:
value := data[key][0] value := data[key]
// Hash each key and value: // Hash each key and value:
fmt.Fprintf(hash, "key=%s :: value=%s\n", key, 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) result = url.Values(data)
// Append the sign: // Append the sign:
result.Add(`checksum`, strings.ToUpper(fmt.Sprintf("%x", hash.Sum(nil)))) result.Add(`checksum`, fmt.Sprintf("%x", hash.Sum(nil)))
return return
} }
-1
View File
@@ -8,5 +8,4 @@ type ICCCGetHosts struct {
type ICCCGetHostsAnswer struct { type ICCCGetHostsAnswer struct {
Hostnames []string Hostnames []string
IPAddressesPorts []string IPAddressesPorts []string
Kinds []byte
} }
-1
View File
@@ -9,5 +9,4 @@ type ICCCGetListenersAnswer struct {
Channels []string Channels []string
Commands []string Commands []string
IPAddressesPorts []string IPAddressesPorts []string
Kinds []byte
} }
+1 -1
View File
@@ -6,7 +6,7 @@ type ICCCGetVersion struct {
// Answer to the version request // Answer to the version request
type ICCCGetVersionAnswer struct { type ICCCGetVersionAnswer struct {
Kind byte // Ocean || Component Kind string // Ocean || Component
Name string // Ocean: Hostname; Components: Name Name string // Ocean: Hostname; Components: Name
Version string // The current version Version string // The current version
} }
@@ -6,5 +6,4 @@ type ICCCListenerUpdate struct {
Command string Command string
IPAddressPort string IPAddressPort string
IsActiveNew bool IsActiveNew bool
Kind byte
} }
-1
View File
@@ -4,5 +4,4 @@ package SystemMessages
type ICCCRegisterHost struct { type ICCCRegisterHost struct {
Hostname string // The hostname for the end-point Hostname string // The hostname for the end-point
IPAddressPort string // The IP address and port 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 Command string // The provided command
IsActive bool // Is the command active? IsActive bool // Is the command active?
IPAddressPort string // The IP address and port for the end-point IPAddressPort string // The IP address and port for the end-point
Kind byte // Ocean || Component
} }
-3
View File
@@ -16,9 +16,6 @@ const (
ChannelICCC string = `System::ICCC` // A common ICCC channel. ChannelICCC string = `System::ICCC` // A common ICCC channel.
ChannelPING string = `System::Ping` // A channel for pings. ChannelPING string = `System::Ping` // A channel for pings.
ChannelLOGGING string = `System::Logging` // A channel for send log events to the logging system ChannelLOGGING string = `System::Logging` // A channel for send log events to the logging system
KindOCEAN byte = 0x0
KindComponent byte = 0x1
KindALL byte = 0xFF
) )
var ( var (
+2 -8
View File
@@ -7,7 +7,7 @@ import (
) )
// Function to broadcast a message to all listeners. // 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() cacheListenerDatabaseLock.RLock()
defer cacheListenerDatabaseLock.RUnlock() defer cacheListenerDatabaseLock.RUnlock()
@@ -21,16 +21,10 @@ func WriteMessage2All(channel, command string, kind byte, message interface{}, a
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() { for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
listener := entry.Value.(Scheme.Listener) listener := entry.Value.(Scheme.Listener)
// If the channel, command and kind matches, deliver the message: // If the channel and the command matches, deliver the message:
if kind == KindALL {
if listener.Channel == channel && listener.Command == command { if listener.Channel == channel && listener.Command == command {
matchingListener = append(matchingListener, listener) matchingListener = append(matchingListener, listener)
} }
} else {
if listener.Channel == channel && listener.Command == command && listener.Kind == kind {
matchingListener = append(matchingListener, listener)
}
}
} }
// Was not able to find any matching listener? // Was not able to find any matching listener?
+2 -9
View File
@@ -8,7 +8,7 @@ import (
) )
// Function to write a message to any listener. // 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() cacheListenerDatabaseLock.RLock()
defer cacheListenerDatabaseLock.RUnlock() defer cacheListenerDatabaseLock.RUnlock()
@@ -22,18 +22,11 @@ func WriteMessage2Any(channel, command string, kind byte, message interface{}, a
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() { for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
listener := entry.Value.(Scheme.Listener) listener := entry.Value.(Scheme.Listener)
// If the channel, command and kind matches, store the listener: // If the channel and the command matches, store the listener:
if kind == KindALL {
if listener.Channel == channel && listener.Command == command { if listener.Channel == channel && listener.Command == command {
entries = entries[:len(entries)+1] entries = entries[:len(entries)+1]
entries[counter] = listener entries[counter] = listener
} }
} else {
if listener.Channel == channel && listener.Command == command && listener.Kind == kind {
entries = entries[:len(entries)+1]
entries[counter] = listener
}
}
} }
count := len(entries) count := len(entries)
-7
View File
@@ -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
View File
@@ -121,16 +121,10 @@ func clearEntry(entry Meta.Entry) (result Meta.Entry) {
// Remove white spaces from the parameters. // Remove white spaces from the parameters.
func clearParameters(oldParameters []string) (result []string) { func clearParameters(oldParameters []string) (result []string) {
for n := 0; n < len(oldParameters); n++ { for n := 0; n < len(oldParameters); n++ {
// Remove white spaces:
oldParameters[n] = removeWhitespaces(oldParameters[n]) 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 return
} }
-1
View File
@@ -30,7 +30,6 @@ func ICCCNewLogEventReceiver(data map[string][]string) (result map[string][]stri
// Provide a log entry: // Provide a log entry:
logEntry := LM.Entry{} logEntry := LM.Entry{}
logEntry.Project = Log.GetProjectName()
logEntry.Sender = LM.Sender(messageData.Sender) logEntry.Sender = LM.Sender(messageData.Sender)
logEntry.Category = LM.ParseCategory(messageData.Category) logEntry.Category = LM.ParseCategory(messageData.Category)
logEntry.Level = LM.ParseLevel(messageData.Level) logEntry.Level = LM.ParseLevel(messageData.Level)
BIN
View File
Binary file not shown.
@@ -29,7 +29,7 @@ func icccComponentStartUpMessageReceiver(data map[string][]string) (result map[s
messageData := obj.(SystemMessages.ICCCComponentStartUpMessage) messageData := obj.(SystemMessages.ICCCComponentStartUpMessage)
// Provide a log entry: // Provide a log entry:
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: The external component is now up and ready.`, fmt.Sprintf("ipAddressPort='%s'", messageData.IPAddressPort), fmt.Sprintf("name='%s'", messageData.Name), fmt.Sprintf("version='%s'", messageData.Version)) Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: The external component is now up and ready.`, fmt.Sprintf("ipAddressPort=%s", messageData.IPAddressPort), fmt.Sprintf("name=", messageData.Name), fmt.Sprintf("version=", messageData.Version))
// An answer is necessary: // An answer is necessary:
return ICCC.Message2Data("", "", SystemMessages.AnswerACK) return ICCC.Message2Data("", "", SystemMessages.AnswerACK)
+1 -1
View File
@@ -29,7 +29,7 @@ func icccOceanStartUpMessageReceiver(data map[string][]string) (result map[strin
messageData := obj.(SystemMessages.ICCCOceanStartUpMessage) messageData := obj.(SystemMessages.ICCCOceanStartUpMessage)
// Provide a log entry: // Provide a log entry:
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: The Ocean server is now up and ready.`, fmt.Sprintf("public server='%s'", messageData.PublicIPAddressPort), fmt.Sprintf("admin server='%s'", messageData.AdminIPAddressPort), fmt.Sprintf("Ocean's version='%s'", messageData.OceanVersion)) Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `ICCC message: The Ocean server is now up and ready.`, fmt.Sprintf("public server=%s", messageData.PublicIPAddressPort), fmt.Sprintf("admin server=%s", messageData.AdminIPAddressPort), fmt.Sprintf("Ocean's version=%s", messageData.OceanVersion))
// An answer is necessary: // An answer is necessary:
return ICCC.Message2Data("", "", SystemMessages.AnswerACK) return ICCC.Message2Data("", "", SystemMessages.AnswerACK)
+1 -1
View File
@@ -1,5 +1,5 @@
package Version package Version
var ( var (
oceansVersion string = `2.0.5` // Ocean's current version oceansVersion string = `2.0.0` // Ocean's current version
) )
+1 -1
View File
@@ -30,7 +30,7 @@ func Start() {
} }
// Notify the whole cluster, that this server is now up and ready: // 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 { for n, obj := range answers {
if obj != nil { if obj != nil {
answer := obj.(SystemMessages.ICCCDefaultAnswer) answer := obj.(SystemMessages.ICCCDefaultAnswer)