Initial commit of Ocean's local development
This commit is contained in:
Thorsten Sommer committed
1 parent
9944a9e1df
commit
86451938ec
116 files changed
+3320
No files matched your search
@@ -0,0 +1,122 @@
|
||||
package ICCC
|
||||
|
||||
import "fmt"
|
||||
import "reflect"
|
||||
import "strconv"
|
||||
|
||||
func Data2Message(target interface{}, data map[string][]string) (channel, command string, obj interface{}) {
|
||||
if data == nil || len(data) == 0 {
|
||||
channel = ``
|
||||
command = ``
|
||||
obj = nil
|
||||
return
|
||||
}
|
||||
|
||||
element := reflect.ValueOf(target)
|
||||
element = element.Elem()
|
||||
elementType := element.Type()
|
||||
|
||||
channel = data[`channel`][0]
|
||||
command = data[`command`][0]
|
||||
for i := 0; i < element.NumField(); i++ {
|
||||
field := element.Field(i)
|
||||
switch field.Kind().String() {
|
||||
|
||||
case `int64`:
|
||||
mapName := fmt.Sprintf(`int:%s`, elementType.Field(i).Name)
|
||||
mapValue := data[mapName][0]
|
||||
v, _ := strconv.ParseInt(mapValue, 10, 64)
|
||||
field.SetInt(v)
|
||||
|
||||
case `string`:
|
||||
mapName := fmt.Sprintf(`str:%s`, elementType.Field(i).Name)
|
||||
mapValue := data[mapName][0]
|
||||
field.SetString(mapValue)
|
||||
|
||||
case `float64`:
|
||||
mapName := fmt.Sprintf(`f64:%s`, elementType.Field(i).Name)
|
||||
mapValue := data[mapName][0]
|
||||
v, _ := strconv.ParseFloat(mapValue, 64)
|
||||
field.SetFloat(v)
|
||||
|
||||
case `bool`:
|
||||
mapName := fmt.Sprintf(`bool:%s`, elementType.Field(i).Name)
|
||||
mapValue := data[mapName][0]
|
||||
v, _ := strconv.ParseBool(mapValue)
|
||||
field.SetBool(v)
|
||||
|
||||
case `uint8`:
|
||||
mapName := fmt.Sprintf(`ui8:%s`, elementType.Field(i).Name)
|
||||
mapValue := data[mapName][0]
|
||||
v, _ := strconv.ParseUint(mapValue, 16, 8)
|
||||
field.SetUint(v)
|
||||
|
||||
case `slice`:
|
||||
sliceInterface := field.Interface()
|
||||
sliceKind := reflect.ValueOf(sliceInterface).Type().String()
|
||||
|
||||
switch sliceKind {
|
||||
case `[]uint8`: // bytes
|
||||
mapName := fmt.Sprintf(`ui8[]:%s`, elementType.Field(i).Name)
|
||||
mapValues := data[mapName]
|
||||
fieldLen := len(mapValues)
|
||||
fieldData := make([]uint8, fieldLen, fieldLen)
|
||||
for n, mapValue := range mapValues {
|
||||
v, _ := strconv.ParseUint(mapValue, 16, 8)
|
||||
fieldData[n] = byte(v)
|
||||
}
|
||||
|
||||
fieldDataValue := reflect.ValueOf(fieldData)
|
||||
field.Set(fieldDataValue)
|
||||
|
||||
case `[]int64`:
|
||||
mapName := fmt.Sprintf(`int[]:%s`, elementType.Field(i).Name)
|
||||
mapValues := data[mapName]
|
||||
fieldLen := len(mapValues)
|
||||
fieldData := make([]int64, fieldLen, fieldLen)
|
||||
for n, mapValue := range mapValues {
|
||||
v, _ := strconv.ParseInt(mapValue, 10, 64)
|
||||
fieldData[n] = v
|
||||
}
|
||||
|
||||
fieldDataValue := reflect.ValueOf(fieldData)
|
||||
field.Set(fieldDataValue)
|
||||
|
||||
case `[]bool`:
|
||||
mapName := fmt.Sprintf(`bool[]:%s`, elementType.Field(i).Name)
|
||||
mapValues := data[mapName]
|
||||
fieldLen := len(mapValues)
|
||||
fieldData := make([]bool, fieldLen, fieldLen)
|
||||
for n, mapValue := range mapValues {
|
||||
v, _ := strconv.ParseBool(mapValue)
|
||||
fieldData[n] = v
|
||||
}
|
||||
|
||||
fieldDataValue := reflect.ValueOf(fieldData)
|
||||
field.Set(fieldDataValue)
|
||||
|
||||
case `[]string`:
|
||||
mapName := fmt.Sprintf(`str[]:%s`, elementType.Field(i).Name)
|
||||
mapValues := data[mapName]
|
||||
fieldDataValue := reflect.ValueOf(mapValues)
|
||||
field.Set(fieldDataValue)
|
||||
|
||||
case `[]float64`:
|
||||
mapName := fmt.Sprintf(`f64[]:%s`, elementType.Field(i).Name)
|
||||
mapValues := data[mapName]
|
||||
fieldLen := len(mapValues)
|
||||
fieldData := make([]float64, fieldLen, fieldLen)
|
||||
for n, mapValue := range mapValues {
|
||||
v, _ := strconv.ParseFloat(mapValue, 64)
|
||||
fieldData[n] = v
|
||||
}
|
||||
|
||||
fieldDataValue := reflect.ValueOf(fieldData)
|
||||
field.Set(fieldDataValue)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
obj = target
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
/*
|
||||
This is the "[I]nter Data [C]enter Se[c]ure [C]ommunication" :)
|
||||
*/
|
||||
package ICCC
|
||||
@@ -0,0 +1,34 @@
|
||||
package ICCC
|
||||
|
||||
import "fmt"
|
||||
import "net/http"
|
||||
import "github.com/SommerEngineering/Ocean/Tools"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func ICCCHandler(response http.ResponseWriter, request *http.Request) {
|
||||
if errParse := request.ParseForm(); errParse != nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityCritical, LM.ImpactCritical, LM.MessageNameNETWORK, `Was not able to parse the HTTP form data from an ICCC message!`)
|
||||
http.NotFound(response, request)
|
||||
return
|
||||
}
|
||||
|
||||
messageData := map[string][]string(request.PostForm)
|
||||
channel := messageData[`channel`][0]
|
||||
command := messageData[`command`][0]
|
||||
password := messageData[`InternalCommPassword`][0]
|
||||
|
||||
if password != Tools.InternalCommPassword() {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelSECURITY, LM.SeverityCritical, LM.ImpactNone, LM.MessageNamePASSWORD, `Received a ICCC message with wrong password!`, request.RemoteAddr)
|
||||
http.NotFound(response, request)
|
||||
return
|
||||
}
|
||||
|
||||
key := fmt.Sprintf(`%s::%s`, channel, command)
|
||||
listener := listeners[key]
|
||||
if listener == nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityCritical, LM.ImpactUnknown, LM.MessageNameCONFIGURATION, `Was not able to find the correct listener for these ICCC message.`, `channel=`+channel, `command`+command, `hostname=`+Tools.ThisHostname())
|
||||
} else {
|
||||
listener(messageData)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package ICCC
|
||||
|
||||
import "strings"
|
||||
import "container/list"
|
||||
import "github.com/SommerEngineering/Ocean/Tools"
|
||||
import "github.com/SommerEngineering/Ocean/ConfigurationDB"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func init() {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Start init of ICCC.`)
|
||||
defer Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Done init ICCC.`)
|
||||
|
||||
cacheListenerDatabase = list.New()
|
||||
listeners = make(map[string]func(data map[string][]string))
|
||||
|
||||
allHostsIPAddresses := Tools.ReadAllIPAddresses4ThisHost()
|
||||
oceanHostnameAndPort := ConfigurationDB.Read(`OceanHostnameAndPort`)
|
||||
port := oceanHostnameAndPort[strings.Index(oceanHostnameAndPort, `:`):]
|
||||
correctAddressWithPort = allHostsIPAddresses[0] + port
|
||||
|
||||
initDB()
|
||||
registerHost2Database()
|
||||
initCacheTimer()
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package ICCC
|
||||
|
||||
import "labix.org/v2/mgo"
|
||||
import "github.com/SommerEngineering/Ocean/CustomerDB"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func initDB() {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Start init of the ICCC collections.`)
|
||||
defer Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Done init the ICCC collection.`)
|
||||
|
||||
// Get the database:
|
||||
db = CustomerDB.DB()
|
||||
|
||||
if db == nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityCritical, LM.ImpactCritical, LM.MessageNameDATABASE, `Was not able to get the customer database.`)
|
||||
return
|
||||
}
|
||||
|
||||
// Get my collections:
|
||||
collectionListener = db.C(`ICCCListener`)
|
||||
collectionHosts = db.C(`ICCCHosts`)
|
||||
|
||||
// Take care about the indexes for ICCCListener:
|
||||
collectionListener.EnsureIndexKey(`Command`)
|
||||
collectionListener.EnsureIndexKey(`Command`, `IsActive`)
|
||||
|
||||
collectionListener.EnsureIndexKey(`Command`, `Channel`)
|
||||
collectionListener.EnsureIndexKey(`Command`, `Channel`, `IsActive`)
|
||||
|
||||
collectionListener.EnsureIndexKey(`Channel`)
|
||||
collectionListener.EnsureIndexKey(`Channel`, `IsActive`)
|
||||
collectionListener.EnsureIndexKey(`Channel`, `Command`, `IPAddressPort`, `IsActive`)
|
||||
collectionListener.EnsureIndexKey(`Channel`, `Command`, `IsActive`)
|
||||
|
||||
collectionListener.EnsureIndexKey(`IsActive`)
|
||||
collectionListener.EnsureIndexKey(`IsActive`, `IPAddressPort`)
|
||||
|
||||
collectionListener.EnsureIndexKey(`IPAddressPort`)
|
||||
|
||||
indexName1 := mgo.Index{}
|
||||
indexName1.Key = []string{`Channel`, `Command`, `IPAddressPort`}
|
||||
indexName1.Unique = true
|
||||
collectionListener.EnsureIndex(indexName1)
|
||||
|
||||
// Index for hosts:
|
||||
collectionHosts.EnsureIndexKey(`Hostname`, `IPAddressPort`)
|
||||
|
||||
indexName2 := mgo.Index{}
|
||||
indexName2.Key = []string{`Hostname`}
|
||||
indexName2.Unique = true
|
||||
collectionHosts.EnsureIndex(indexName2)
|
||||
|
||||
indexName3 := mgo.Index{}
|
||||
indexName3.Key = []string{`IPAddressPort`}
|
||||
indexName3.Unique = true
|
||||
collectionHosts.EnsureIndex(indexName3)
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package ICCC
|
||||
|
||||
import "fmt"
|
||||
import "time"
|
||||
import "labix.org/v2/mgo/bson"
|
||||
import "github.com/SommerEngineering/Ocean/Shutdown"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func initCacheTimer() {
|
||||
|
||||
go func() {
|
||||
for {
|
||||
|
||||
if Shutdown.IsDown() {
|
||||
return
|
||||
}
|
||||
|
||||
lastCount := cacheListenerDatabase.Len()
|
||||
selection := bson.D{{`IsActive`, true}}
|
||||
entriesIterator := collectionListener.Find(selection).Iter()
|
||||
entry := Scheme.Listener{}
|
||||
|
||||
cacheListenerDatabaseLock.Lock()
|
||||
cacheListenerDatabase.Init()
|
||||
for entriesIterator.Next(&entry) {
|
||||
cacheListenerDatabase.PushBack(entry)
|
||||
}
|
||||
|
||||
cacheListenerDatabaseLock.Unlock()
|
||||
nextDuration := time.Duration(5) * time.Minute
|
||||
if cacheListenerDatabase.Len() == 0 {
|
||||
nextDuration = time.Duration(10) * time.Second
|
||||
}
|
||||
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameEXECUTE, `The listener cache was refreshed with the values from the database.`, fmt.Sprintf(`last count=%d`, lastCount), fmt.Sprintf(`new count=%d`, cacheListenerDatabase.Len()))
|
||||
time.Sleep(nextDuration)
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package ICCC
|
||||
|
||||
import "fmt"
|
||||
import "reflect"
|
||||
import "strconv"
|
||||
|
||||
func message2Data(channel, command string, message interface{}) (data map[string][]string) {
|
||||
if message == nil {
|
||||
data = make(map[string][]string)
|
||||
return
|
||||
}
|
||||
|
||||
element := reflect.ValueOf(message)
|
||||
elementType := element.Type()
|
||||
|
||||
data = make(map[string][]string)
|
||||
data[`command`] = []string{command}
|
||||
data[`channel`] = []string{channel}
|
||||
|
||||
for i := 0; i < element.NumField(); i++ {
|
||||
field := element.Field(i)
|
||||
keyName := elementType.Field(i).Name
|
||||
|
||||
switch field.Kind().String() {
|
||||
|
||||
case `int64`:
|
||||
key := fmt.Sprintf(`int:%s`, keyName)
|
||||
data[key] = []string{strconv.FormatInt(field.Int(), 10)}
|
||||
|
||||
case `string`:
|
||||
key := fmt.Sprintf(`str:%s`, keyName)
|
||||
data[key] = []string{field.String()}
|
||||
|
||||
case `float64`:
|
||||
key := fmt.Sprintf(`f64:%s`, keyName)
|
||||
data[key] = []string{strconv.FormatFloat(field.Float(), 'f', 9, 64)}
|
||||
|
||||
case `bool`:
|
||||
key := fmt.Sprintf(`bool:%s`, keyName)
|
||||
data[key] = []string{strconv.FormatBool(field.Bool())}
|
||||
|
||||
case `uint8`: // byte
|
||||
key := fmt.Sprintf(`ui8:%s`, keyName)
|
||||
data[key] = []string{strconv.FormatUint(field.Uint(), 16)}
|
||||
|
||||
case `slice`:
|
||||
sliceLen := field.Len()
|
||||
if sliceLen > 0 {
|
||||
sliceKind := field.Index(0).Kind()
|
||||
key := ``
|
||||
dataValues := make([]string, sliceLen, sliceLen)
|
||||
switch sliceKind.String() {
|
||||
case `uint8`: // bytes
|
||||
key = fmt.Sprintf(`ui8[]:%s`, keyName)
|
||||
values := field.Interface().([]uint8)
|
||||
for index, value := range values {
|
||||
dataValues[index] = strconv.FormatUint(uint64(value), 16)
|
||||
}
|
||||
|
||||
case `int64`:
|
||||
key = fmt.Sprintf(`int[]:%s`, keyName)
|
||||
values := field.Interface().([]int64)
|
||||
for index, value := range values {
|
||||
dataValues[index] = strconv.FormatInt(value, 10)
|
||||
}
|
||||
|
||||
case `bool`:
|
||||
key = fmt.Sprintf(`bool[]:%s`, keyName)
|
||||
values := field.Interface().([]bool)
|
||||
for index, value := range values {
|
||||
dataValues[index] = strconv.FormatBool(value)
|
||||
}
|
||||
|
||||
case `string`:
|
||||
key = fmt.Sprintf(`str[]:%s`, keyName)
|
||||
values := field.Interface().([]string)
|
||||
for index, value := range values {
|
||||
dataValues[index] = value
|
||||
}
|
||||
|
||||
case `float64`:
|
||||
key = fmt.Sprintf(`f64[]:%s`, keyName)
|
||||
values := field.Interface().([]float64)
|
||||
for index, value := range values {
|
||||
dataValues[index] = strconv.FormatFloat(value, 'f', 9, 64)
|
||||
}
|
||||
}
|
||||
|
||||
data[key] = dataValues
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
package ICCC
|
||||
|
||||
import "labix.org/v2/mgo/bson"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func register2Database(channel, command string) {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `Register this ICCC command in to the database.`, `channel=`+channel, `command=`+command)
|
||||
|
||||
// Case: Exist and active :)
|
||||
emptyEntry := Scheme.Listener{}
|
||||
selection := bson.D{{`Channel`, channel}, {`Command`, command}, {`IPAddressPort`, correctAddressWithPort}, {`IsActive`, true}}
|
||||
count1, _ := collectionListener.Find(selection).Count()
|
||||
|
||||
if count1 == 1 {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.SeverityHigh, LM.ImpactHigh, LM.MessageNameSTARTUP, `This ICCC command is already known and active.`, `Please shutdown the system next time!`)
|
||||
return
|
||||
}
|
||||
|
||||
// Case: Exist but not active
|
||||
selection = bson.D{{`Channel`, channel}, {`Command`, command}, {`IPAddressPort`, correctAddressWithPort}, {`IsActive`, false}}
|
||||
notActiveEntry := Scheme.Listener{}
|
||||
collectionListener.Find(selection).One(¬ActiveEntry)
|
||||
|
||||
if notActiveEntry != emptyEntry {
|
||||
notActiveEntry.IsActive = true
|
||||
collectionListener.Update(selection, notActiveEntry)
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.SeverityNone, LM.ImpactNone, LM.MessageNameSTARTUP, `This ICCC command is already known but it was not active.`, `The command is active now!`)
|
||||
return
|
||||
}
|
||||
|
||||
// Case: Not exist
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityCritical, LM.ImpactNone, LM.MessageNameCONFIGURATION, `This ICCC command is not known.`, `Create now a new entry!`)
|
||||
|
||||
entry := Scheme.Listener{}
|
||||
entry.Channel = channel
|
||||
entry.Command = command
|
||||
entry.IsActive = true
|
||||
entry.IPAddressPort = correctAddressWithPort
|
||||
|
||||
if err := collectionListener.Insert(entry); err != nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityCritical, LM.ImpactCritical, LM.MessageNameDATABASE, `It was not possible to add this ICCC command!`, err.Error(), `channel=`+channel, `command=`+command)
|
||||
} else {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameCONFIGURATION, `This ICCC command is now known and active.`)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package ICCC
|
||||
|
||||
import "labix.org/v2/mgo/bson"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Tools"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func registerHost2Database() {
|
||||
host := Scheme.Host{}
|
||||
host.Hostname = Tools.ThisHostname()
|
||||
host.IPAddressPort = correctAddressWithPort
|
||||
|
||||
selection := bson.D{{`Hostname`, host.Hostname}, {`IPAddressPort`, host.IPAddressPort}}
|
||||
count, _ := collectionHosts.Find(selection).Count()
|
||||
|
||||
if count == 1 {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameCONFIGURATION, `This host is already registered!`, `host=`+host.Hostname, `address=`+host.IPAddressPort)
|
||||
} else {
|
||||
if errInsert := collectionHosts.Insert(host); errInsert != nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityCritical, LM.ImpactCritical, LM.MessageNameDATABASE, `Was not able to register this host.`, errInsert.Error(), `host=`+host.Hostname, `address=`+host.IPAddressPort)
|
||||
} else {
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `This host is now registered.`, `host=`+host.Hostname, `address=`+host.IPAddressPort)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package ICCC
|
||||
|
||||
import "fmt"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func Registrar(channel, command string, callback func(data map[string][]string)) {
|
||||
listenersLock.Lock()
|
||||
defer listenersLock.Unlock()
|
||||
|
||||
register2Database(channel, command)
|
||||
listeners[fmt.Sprintf(`%s::%s`, channel, command)] = callback
|
||||
|
||||
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameCONFIGURATION, `The registrar has registered a new ICCC command.`, `channel=`+channel, `command=`+command)
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
package Scheme
|
||||
|
||||
type Host struct {
|
||||
Hostname string `bson:"Hostname"`
|
||||
IPAddressPort string `bson:"IPAddressPort"`
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package Scheme
|
||||
|
||||
type Listener struct {
|
||||
Channel string `bson:"Channel"`
|
||||
Command string `bson:"Command"`
|
||||
IsActive bool `bson:"IsActive"`
|
||||
IPAddressPort string `bson:"IPAddressPort"`
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package ICCC
|
||||
|
||||
import "net/http"
|
||||
import "net/url"
|
||||
import "github.com/SommerEngineering/Ocean/Tools"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func sendMessage(listener Scheme.Listener, data map[string][]string) {
|
||||
|
||||
valuesHTTP := url.Values(data)
|
||||
valuesHTTP.Add(`InternalCommPassword`, Tools.InternalCommPassword())
|
||||
if _, err := http.PostForm(`http://`+listener.IPAddressPort+`/ICCC`, valuesHTTP); err != nil {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityCritical, LM.ImpactUnknown, LM.MessageNameNETWORK, `Was not able to send the ICCC message.`, err.Error())
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package ICCC
|
||||
|
||||
import "labix.org/v2/mgo/bson"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
/*
|
||||
Please do not use this type. It is an internal type of Ocean to provide a shutdown function!
|
||||
*/
|
||||
type ShutdownFunction struct {
|
||||
}
|
||||
|
||||
/*
|
||||
This function is called if the Ocean server is shutting down.
|
||||
*/
|
||||
func (a ShutdownFunction) Shutdown() {
|
||||
Log.LogShort(senderName, LM.CategoryAPP, LM.LevelWARN, LM.MessageNameSHUTDOWN, `Shutting down now all ICCC listener for this host.`)
|
||||
|
||||
selection := bson.D{{`IPAddressPort`, correctAddressWithPort}}
|
||||
entry := Scheme.Listener{}
|
||||
iterator := collectionListener.Find(selection).Iter()
|
||||
for iterator.Next(&entry) {
|
||||
selectionUpdate := bson.D{{`Channel`, entry.Channel}, {`Command`, entry.Command}, {`IPAddressPort`, correctAddressWithPort}}
|
||||
entry.IsActive = false
|
||||
collectionListener.Update(selectionUpdate, entry)
|
||||
}
|
||||
|
||||
Log.LogShort(senderName, LM.CategoryAPP, LM.LevelWARN, LM.MessageNameSHUTDOWN, `Done shutting down now all ICCC listener for this host.`)
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package ICCC
|
||||
|
||||
import "sync"
|
||||
import "container/list"
|
||||
import "labix.org/v2/mgo"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
const (
|
||||
ChannelSYSTEM string = `System`
|
||||
ChannelNUMGEN string = `System::NumGen`
|
||||
ChannelSHUTDOWN string = `System::Shutdown`
|
||||
ChannelSTARTUP string = `System::Startup`
|
||||
ChannelICCC string = `System::ICCC`
|
||||
)
|
||||
|
||||
var (
|
||||
senderName LM.Sender = `ICCC`
|
||||
db *mgo.Database = nil
|
||||
collectionListener *mgo.Collection = nil
|
||||
collectionHosts *mgo.Collection = nil
|
||||
reservedSystemChannels []string = []string{ChannelSYSTEM, ChannelNUMGEN, ChannelSHUTDOWN, ChannelSTARTUP, ChannelICCC}
|
||||
listeners map[string]func(data map[string][]string) = nil
|
||||
listenersLock sync.RWMutex = sync.RWMutex{}
|
||||
cacheListenerDatabase *list.List = nil
|
||||
cacheListenerDatabaseLock sync.RWMutex = sync.RWMutex{}
|
||||
correctAddressWithPort string = ``
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
package ICCC
|
||||
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func WriteMessage2All(channel, command string, message interface{}) {
|
||||
cacheListenerDatabaseLock.RLock()
|
||||
defer cacheListenerDatabaseLock.RUnlock()
|
||||
|
||||
data := message2Data(channel, command, message)
|
||||
counter := 0
|
||||
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
|
||||
listener := entry.Value.(Scheme.Listener)
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
go sendMessage(listener, data)
|
||||
counter++
|
||||
}
|
||||
}
|
||||
|
||||
if counter == 0 {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityCritical, LM.ImpactUnknown, LM.MessageNameCONFIGURATION, `It was not able to deliver this message, because no listener was found!`, `channel=`+channel, `command=`+command)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package ICCC
|
||||
|
||||
import "github.com/SommerEngineering/Ocean/Tools"
|
||||
import "github.com/SommerEngineering/Ocean/ICCC/Scheme"
|
||||
import "github.com/SommerEngineering/Ocean/Log"
|
||||
import LM "github.com/SommerEngineering/Ocean/Log/Meta"
|
||||
|
||||
func WriteMessage2Any(channel, command string, message interface{}) {
|
||||
cacheListenerDatabaseLock.RLock()
|
||||
defer cacheListenerDatabaseLock.RUnlock()
|
||||
|
||||
data := message2Data(channel, command, message)
|
||||
maxCount := cacheListenerDatabase.Len()
|
||||
entries := make([]Scheme.Listener, 0, maxCount)
|
||||
counter := 0
|
||||
for entry := cacheListenerDatabase.Front(); entry != nil; entry = entry.Next() {
|
||||
listener := entry.Value.(Scheme.Listener)
|
||||
if listener.Channel == channel && listener.Command == command {
|
||||
entries = entries[:len(entries)+1]
|
||||
entries[counter] = listener
|
||||
}
|
||||
}
|
||||
|
||||
count := len(entries)
|
||||
if count > 0 {
|
||||
listener := entries[Tools.RandomInteger(count)]
|
||||
go sendMessage(listener, data)
|
||||
} else {
|
||||
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityCritical, LM.ImpactUnknown, LM.MessageNameCONFIGURATION, `It was not able to deliver this message to any listener, because no listener was found!`, `channel=`+channel, `command=`+command)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user