Skip to content

GitLab

Commit f9cef090 authored by Sonja Happ's avatar Sonja Happ
Browse files

merge amqp module into infrastructure-component module to avoid circular deps,... 

merge amqp module into infrastructure-component module to avoid circular deps, add TODOs in IC endpoints
parent 4922d8c4
Hide whitespace changes
Inline Side-by-side
amqp/amqp_endpoints.go deleted 100644 → 0
View file @ 4922d8c4
/** AMQP package, endpoints.
*
* @author Sonja Happ <sonja.happ@eonerc.rwth-aachen.de>
* @copyright 2014-2019, Institute for Automation of Complex Power Systems, EONERC
* @license GNU General Public License (version 3)
*
* VILLASweb-backend-go
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*********************************************************************************/
package amqp
import (
"git.rwth-aachen.de/acs/public/villas/web-backend-go/database"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/helper"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/routes/infrastructure-component"
"github.com/gin-gonic/gin"
"log"
"net/http"
)
func RegisterAMQPEndpoint(r *gin.RouterGroup) {
r.POST("/:ICID/action", sendActionToIC)
}
// sendActionToIC godoc
// @Summary Send an action to IC (only available if backend server is started with -amqp parameter)
// @ID sendActionToIC
// @Tags AMQP
// @Produce json
// @Param inputAction query string true "Action for IC"
// @Success 200 {object} docs.ResponseError "Action sent successfully"
// @Failure 400 {object} docs.ResponseError "Bad request"
// @Failure 404 {object} docs.ResponseError "Not found"
// @Failure 422 {object} docs.ResponseError "Unprocessable entity"
// @Failure 500 {object} docs.ResponseError "Internal server error"
// @Param ICID path int true "InfrastructureComponent ID"
// @Router /ic/{ICID}/action [post]
// @Security Bearer
func sendActionToIC(c *gin.Context) {
ok, s := infrastructure_component.CheckPermissions(c, database.ModelInfrastructureComponentAction, database.Update, true)
if !ok {
return
}
var actions []Action
err := c.BindJSON(&actions)
if err != nil {
helper.BadRequestError(c, "Error binding form data to JSON: "+err.Error())
return
}
//now := time.Now()
log.Println("AMQP: Will attempt to send the following actions:", actions)
for _, action := range actions {
/*if action.When == 0 {
action.When = float32(now.Unix())
}*/
action.UUID = new(string)
*action.UUID = s.UUID
err = SendActionAMQP(action)
if err != nil {
helper.InternalServerError(c, "Unable to send actions to IC: "+err.Error())
return
}
}
c.JSON(http.StatusOK, gin.H{
"success": true,
"message": "OK.",
})
}
routes/healthz/healthz_endpoint.go
View file @ f9cef090
......@@ -22,10 +22,10 @@
package healthz
import (
"git.rwth-aachen.de/acs/public/villas/web-backend-go/amqp"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/configuration"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/database"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/helper"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/routes/infrastructure-component"
"github.com/gin-gonic/gin"
"log"
"net/http"
......@@ -68,7 +68,7 @@ func getHealth(c *gin.Context) {
}
if len(url) != 0 {
err = amqp.CheckConnection()
err = infrastructure_component.CheckConnection()
if err != nil {
log.Println(err.Error())
c.JSON(http.StatusInternalServerError, gin.H{
......
routes/healthz/healthz_test.go
View file @ f9cef090
......@@ -22,10 +22,10 @@
package healthz
import (
"git.rwth-aachen.de/acs/public/villas/web-backend-go/amqp"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/configuration"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/database"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/helper"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/routes/infrastructure-component"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
"log"
......@@ -78,7 +78,7 @@ func TestHealthz(t *testing.T) {
amqpURI := "amqp://" + user + ":" + pass + "@" + host
log.Println("AMQP URI is", amqpURI)
err = amqp.ConnectAMQP(amqpURI)
err = infrastructure_component.ConnectAMQP(amqpURI)
assert.NoError(t, err)
// test healthz endpoint for connected DB and AMQP client
......
amqp/amqpclient.go → routes/infrastructure-component/amqpclient.go
View file @ f9cef090
......@@ -19,12 +19,11 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*********************************************************************************/
package amqp
package infrastructure_component
import (
"encoding/json"
"fmt"
infrastructure_component "git.rwth-aachen.de/acs/public/villas/web-backend-go/routes/infrastructure-component"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/jinzhu/gorm"
......@@ -249,129 +248,19 @@ func processMessage(message amqp.Delivery) error {
return fmt.Errorf("AMQP: UUID not valid: %v, message ignored: %v \n", ICUUID, string(message.Body))
}
var sToBeUpdated infrastructure_component.InfrastructureComponent
var sToBeUpdated InfrastructureComponent
err = sToBeUpdated.ByUUID(ICUUID)
if err == gorm.ErrRecordNotFound {
// create new record
err = createNewIC(payload)
err = createNewICviaAMQP(payload)
} else if err != nil {
// database error
err = fmt.Errorf("AMQP: Database error for IC %v DB error message: %v", ICUUID, err)
} else {
// update record
err = updateIC(payload, sToBeUpdated)
// update record based on payload
err = sToBeUpdated.updateICviaAMQP(payload)
}
return err
}
func createNewIC(payload ICUpdate) error {
var newICReq infrastructure_component.AddICRequest
newICReq.InfrastructureComponent.UUID = payload.Properties.UUID
if payload.Properties.Name == nil ||
payload.Properties.Category == nil ||
payload.Properties.Type == nil {
// cannot create new IC because required information (name, type, and/or category missing)
return fmt.Errorf("AMQP: Cannot create new IC, required field(s) is/are missing: name, type, category")
}
newICReq.InfrastructureComponent.Name = *payload.Properties.Name
newICReq.InfrastructureComponent.Category = *payload.Properties.Category
newICReq.InfrastructureComponent.Type = *payload.Properties.Type
// add optional params
if payload.State != nil {
newICReq.InfrastructureComponent.State = *payload.State
} else {
newICReq.InfrastructureComponent.State = "unknown"
}
if payload.Properties.WS_url != nil {
newICReq.InfrastructureComponent.WebsocketURL = *payload.Properties.WS_url
}
if payload.Properties.API_url != nil {
newICReq.InfrastructureComponent.APIURL = *payload.Properties.API_url
}
if payload.Properties.Location != nil {
newICReq.InfrastructureComponent.Location = *payload.Properties.Location
}
if payload.Properties.Description != nil {
newICReq.InfrastructureComponent.Description = *payload.Properties.Description
}
// TODO add JSON start parameter scheme
// set managed externally to true because this IC is created via AMQP
newICReq.InfrastructureComponent.ManagedExternally = newTrue()
// Validate the new IC
err := newICReq.Validate()
if err != nil {
return fmt.Errorf("AMQP: Validation of new IC failed: %v", err)
}
// Create the new IC
newIC := newICReq.CreateIC()
// save IC
err = newIC.Save()
if err != nil {
return fmt.Errorf("AMQP: Saving new IC to DB failed: %v", err)
}
return nil
}
func updateIC(payload ICUpdate, sToBeUpdated infrastructure_component.InfrastructureComponent) error {
var updatedICReq infrastructure_component.UpdateICRequest
if payload.State != nil {
updatedICReq.InfrastructureComponent.State = *payload.State
}
if payload.Properties.Type != nil {
updatedICReq.InfrastructureComponent.Type = *payload.Properties.Type
}
if payload.Properties.Category != nil {
updatedICReq.InfrastructureComponent.Category = *payload.Properties.Category
}
if payload.Properties.Name != nil {
updatedICReq.InfrastructureComponent.Name = *payload.Properties.Name
}
if payload.Properties.WS_url != nil {
updatedICReq.InfrastructureComponent.WebsocketURL = *payload.Properties.WS_url
}
if payload.Properties.API_url != nil {
updatedICReq.InfrastructureComponent.APIURL = *payload.Properties.API_url
}
if payload.Properties.Location != nil {
//postgres.Jsonb{json.RawMessage(`{"location" : " ` + *payload.Properties.Location + `"}`)}
updatedICReq.InfrastructureComponent.Location = *payload.Properties.Location
}
if payload.Properties.Description != nil {
updatedICReq.InfrastructureComponent.Description = *payload.Properties.Description
}
// TODO add JSON start parameter scheme
// set managed externally to true because this IC is updated via AMQP
updatedICReq.InfrastructureComponent.ManagedExternally = newTrue()
// Validate the updated IC
err := updatedICReq.Validate()
if err != nil {
return fmt.Errorf("AMQP: Validation of updated IC failed: %v", err)
}
// Create the updated IC from old IC
updatedIC := updatedICReq.UpdatedIC(sToBeUpdated)
// Finally update the IC in the DB
err = sToBeUpdated.Update(updatedIC)
if err != nil {
return fmt.Errorf("AMQP: Unable to update IC %v in DB: %v", sToBeUpdated.Name, err)
}
return err
}
func newTrue() *bool {
b := true
return &b
}
routes/infrastructure-component/ic_endpoints.go
View file @ f9cef090
......@@ -24,6 +24,7 @@ package infrastructure_component
import (
"git.rwth-aachen.de/acs/public/villas/web-backend-go/helper"
"github.com/gin-gonic/gin"
"log"
"net/http"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/database"
......@@ -38,6 +39,10 @@ func RegisterICEndpoints(r *gin.RouterGroup) {
r.GET("/:ICID/configs", getConfigsOfIC)
}
func RegisterAMQPEndpoint(r *gin.RouterGroup) {
r.POST("/:ICID/action", sendActionToIC)
}
// getICs godoc
// @Summary Get all infrastructure components
// @ID getICs
......@@ -96,6 +101,8 @@ func addIC(c *gin.Context) {
return
}
// TODO add case distinction here for externally managed IC
// Create the new IC from the request
newIC := req.CreateIC()
......@@ -142,6 +149,8 @@ func updateIC(c *gin.Context) {
return
}
// TODO add case distinction here for externally managed IC
// Create the updatedIC from oldIC
updatedIC := req.UpdatedIC(oldIC)
......@@ -196,6 +205,8 @@ func deleteIC(c *gin.Context) {
return
}
// TODO add case distinction here for externally managed IC
// Delete the IC
err := s.delete()
if !helper.DBError(c, err) {
......@@ -231,3 +242,53 @@ func getConfigsOfIC(c *gin.Context) {
}
}
// sendActionToIC godoc
// @Summary Send an action to IC (only available if backend server is started with -amqp parameter)
// @ID sendActionToIC
// @Tags infrastructure-components
// @Produce json
// @Param inputAction query string true "Action for IC"
// @Success 200 {object} docs.ResponseError "Action sent successfully"
// @Failure 400 {object} docs.ResponseError "Bad request"
// @Failure 404 {object} docs.ResponseError "Not found"
// @Failure 422 {object} docs.ResponseError "Unprocessable entity"
// @Failure 500 {object} docs.ResponseError "Internal server error"
// @Param ICID path int true "InfrastructureComponent ID"
// @Router /ic/{ICID}/action [post]
// @Security Bearer
func sendActionToIC(c *gin.Context) {
ok, s := CheckPermissions(c, database.ModelInfrastructureComponentAction, database.Update, true)
if !ok {
return
}
var actions []Action
err := c.BindJSON(&actions)
if err != nil {
helper.BadRequestError(c, "Error binding form data to JSON: "+err.Error())
return
}
//now := time.Now()
log.Println("AMQP: Will attempt to send the following actions:", actions)
for _, action := range actions {
/*if action.When == 0 {
action.When = float32(now.Unix())
}*/
action.UUID = new(string)
*action.UUID = s.UUID
err = SendActionAMQP(action)
if err != nil {
helper.InternalServerError(c, "Unable to send actions to IC: "+err.Error())
return
}
}
c.JSON(http.StatusOK, gin.H{
"success": true,
"message": "OK.",
})
}
routes/infrastructure-component/ic_methods.go
View file @ f9cef090
......@@ -77,3 +77,117 @@ func (s *InfrastructureComponent) getConfigs() ([]database.ComponentConfiguratio
err := db.Order("ID asc").Model(s).Related(&configs, "ComponentConfigurations").Error
return configs, len(configs), err
}
func createNewICviaAMQP(payload ICUpdate) error {
var newICReq AddICRequest
newICReq.InfrastructureComponent.UUID = payload.Properties.UUID
if payload.Properties.Name == nil ||
payload.Properties.Category == nil ||
payload.Properties.Type == nil {
// cannot create new IC because required information (name, type, and/or category missing)
return fmt.Errorf("AMQP: Cannot create new IC, required field(s) is/are missing: name, type, category")
}
newICReq.InfrastructureComponent.Name = *payload.Properties.Name
newICReq.InfrastructureComponent.Category = *payload.Properties.Category
newICReq.InfrastructureComponent.Type = *payload.Properties.Type
// add optional params
if payload.State != nil {
newICReq.InfrastructureComponent.State = *payload.State
} else {
newICReq.InfrastructureComponent.State = "unknown"
}
// TODO check if state is "gone" and abort creation of IC in this case
if payload.Properties.WS_url != nil {
newICReq.InfrastructureComponent.WebsocketURL = *payload.Properties.WS_url
}
if payload.Properties.API_url != nil {
newICReq.InfrastructureComponent.APIURL = *payload.Properties.API_url
}
if payload.Properties.Location != nil {
newICReq.InfrastructureComponent.Location = *payload.Properties.Location
}
if payload.Properties.Description != nil {
newICReq.InfrastructureComponent.Description = *payload.Properties.Description
}
// TODO add JSON start parameter scheme
// set managed externally to true because this IC is created via AMQP
newICReq.InfrastructureComponent.ManagedExternally = newTrue()
// Validate the new IC
err := newICReq.Validate()
if err != nil {
return fmt.Errorf("AMQP: Validation of new IC failed: %v", err)
}
// Create the new IC
newIC := newICReq.CreateIC()
// save IC
err = newIC.Save()
if err != nil {
return fmt.Errorf("AMQP: Saving new IC to DB failed: %v", err)
}
return nil
}
func (s *InfrastructureComponent) updateICviaAMQP(payload ICUpdate) error {
var updatedICReq UpdateICRequest
if payload.State != nil {
updatedICReq.InfrastructureComponent.State = *payload.State
// TODO check if state is "gone" and attempt to remove IC from DB if it still exists
// TODO if state is different from "gone", continue to update the IC
}
if payload.Properties.Type != nil {
updatedICReq.InfrastructureComponent.Type = *payload.Properties.Type
}
if payload.Properties.Category != nil {
updatedICReq.InfrastructureComponent.Category = *payload.Properties.Category
}
if payload.Properties.Name != nil {
updatedICReq.InfrastructureComponent.Name = *payload.Properties.Name
}
if payload.Properties.WS_url != nil {
updatedICReq.InfrastructureComponent.WebsocketURL = *payload.Properties.WS_url
}
if payload.Properties.API_url != nil {
updatedICReq.InfrastructureComponent.APIURL = *payload.Properties.API_url
}
if payload.Properties.Location != nil {
//postgres.Jsonb{json.RawMessage(`{"location" : " ` + *payload.Properties.Location + `"}`)}
updatedICReq.InfrastructureComponent.Location = *payload.Properties.Location
}
if payload.Properties.Description != nil {
updatedICReq.InfrastructureComponent.Description = *payload.Properties.Description
}
// TODO add JSON start parameter scheme
// set managed externally to true because this IC is updated via AMQP
updatedICReq.InfrastructureComponent.ManagedExternally = newTrue()
// Validate the updated IC
err := updatedICReq.Validate()
if err != nil {
return fmt.Errorf("AMQP: Validation of updated IC failed: %v", err)
}
// Create the updated IC from old IC
updatedIC := updatedICReq.UpdatedIC(*s)
// Finally update the IC in the DB
err = s.Update(updatedIC)
if err != nil {
return fmt.Errorf("AMQP: Unable to update IC %v in DB: %v", s.Name, err)
}
return err
}
func newTrue() *bool {
b := true
return &b
}
start.go
View file @ f9cef090
......@@ -23,12 +23,12 @@ package main
import (
"fmt"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/amqp"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/configuration"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/database"
apidocs "git.rwth-aachen.de/acs/public/villas/web-backend-go/doc/api" // doc/api folder is used by Swag CLI, you have to import it
"git.rwth-aachen.de/acs/public/villas/web-backend-go/helper"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/routes"
"git.rwth-aachen.de/acs/public/villas/web-backend-go/routes/infrastructure-component"
"github.com/gin-gonic/gin"
"log"
)
......@@ -104,7 +104,7 @@ func main() {
if amqphost != "" {
// create amqp URL based on username, password and host
amqpurl := "amqp://" + amqpuser + ":" + amqppass + "@" + amqphost
err = amqp.StartAMQP(amqpurl, api)
err = infrastructure_component.StartAMQP(amqpurl, api)
if err != nil {
panic(err)
}
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment