coverage.*
main
dmaapmediatorproducer
+__debug_bin*
)
type Config struct {
- LogLevel string
- JobResultUri string
- InfoCoordinatorAddress string
+ LogLevel string
+ InfoJobCallbackUrl string
+ InfoCoordinatorAddress string
+ InfoProducerSupervisionCallbackUrl string
+}
+
+type ProducerRegistrationInfo struct {
+ InfoProducerSupervisionCallbackUrl string `json:"info_producer_supervision_callback_url"`
+ SupportedInfoTypes []string `json:"supported_info_types"`
+ InfoJobCallbackUrl string `json:"info_job_callback_url"`
}
func New() *Config {
return &Config{
- LogLevel: getEnv("LOG_LEVEL", "Info"),
- JobResultUri: getEnv("JOB_RESULT_URI", ""),
- InfoCoordinatorAddress: getEnv("INFO_COORD_ADDR", "http://enrichmentservice:8083"),
+ LogLevel: getEnv("LOG_LEVEL", "Info"),
+ InfoJobCallbackUrl: getEnv("INFO_JOB_CALLBACK_URL", ""),
+ InfoCoordinatorAddress: getEnv("INFO_COORD_ADDR", "http://enrichmentservice:8083"),
+ InfoProducerSupervisionCallbackUrl: getEnv("INFO_PRODUCER_SUPERVISION_CALLBACK_URL", ""),
}
}
func TestNew_envVarsSetConfigContainSetValues(t *testing.T) {
os.Setenv("LOG_LEVEL", "Debug")
- os.Setenv("JOB_RESULT_URI", "testUrl")
- os.Setenv("INFO_COORD_ADDR", "testAddr")
+ os.Setenv("INFO_JOB_CALLBACK_URL", "jobCallbackUrl")
+ os.Setenv("INFO_COORD_ADDR", "infoCoordAddr")
+ os.Setenv("INFO_PRODUCER_SUPERVISION_CALLBACK_URL", "supervisionCallbackUrl")
defer os.Clearenv()
wantConfig := Config{
- LogLevel: "Debug",
- JobResultUri: "testUrl",
- InfoCoordinatorAddress: "testAddr",
+ LogLevel: "Debug",
+ InfoJobCallbackUrl: "jobCallbackUrl",
+ InfoCoordinatorAddress: "infoCoordAddr",
+ InfoProducerSupervisionCallbackUrl: "supervisionCallbackUrl",
}
if got := New(); !reflect.DeepEqual(got, &wantConfig) {
t.Errorf("New() = %v, want %v", got, &wantConfig)
func TestNew_envVarsNotSetConfigContainDefaultValues(t *testing.T) {
wantConfig := Config{
- LogLevel: "Info",
- JobResultUri: "",
- InfoCoordinatorAddress: "http://enrichmentservice:8083",
+ LogLevel: "Info",
+ InfoJobCallbackUrl: "",
+ InfoCoordinatorAddress: "http://enrichmentservice:8083",
+ InfoProducerSupervisionCallbackUrl: "",
}
if got := New(); !reflect.DeepEqual(got, &wantConfig) {
t.Errorf("New() = %v, want %v", got, &wantConfig)
package config
import (
+ "encoding/json"
"fmt"
"net/url"
)
const registerTypePath = "/data-producer/v1/info-types/"
+const registerProducerPath = "/data-producer/v1/info-producers/"
type Registrator interface {
RegisterTypes(types []*jobtypes.Type) error
+ RegisterProducer(producerId string, producerInfo *ProducerRegistrationInfo)
}
type RegistratorImpl struct {
func (r RegistratorImpl) RegisterTypes(jobTypes []*jobtypes.Type) error {
for _, jobType := range jobTypes {
body := fmt.Sprintf(`{"info_job_data_schema": %v}`, jobType.Schema)
- if error := restclient.Put(r.infoCoordinatorAddress+registerTypePath+url.PathEscape(jobType.Name), []byte(body)); error != nil {
+ if error := restclient.Put(r.infoCoordinatorAddress+registerTypePath+url.PathEscape(jobType.TypeId), []byte(body)); error != nil {
return error
}
log.Debugf("Registered type: %v", jobType)
}
return nil
}
+
+func (r RegistratorImpl) RegisterProducer(producerId string, producerInfo *ProducerRegistrationInfo) error {
+ if body, marshalErr := json.Marshal(producerInfo); marshalErr == nil {
+ if putErr := restclient.Put(r.infoCoordinatorAddress+registerProducerPath+url.PathEscape(producerId), []byte(body)); putErr != nil {
+ return putErr
+ }
+ log.Debugf("Registered producer: %v", producerId)
+ return nil
+ } else {
+ return marshalErr
+ }
+}
restclient.Client = &clientMock
type1 := jobtypes.Type{
- Name: "Type1",
+ TypeId: "Type1",
Schema: `{"title": "Type 1"}`,
}
types := []*jobtypes.Type{&type1}
assertions.Equal(expectedBody, body)
clientMock.AssertNumberOfCalls(t, "Do", 1)
}
+
+func TestRegisterProducer(t *testing.T) {
+ assertions := require.New(t)
+
+ clientMock := mocks.HTTPClient{}
+
+ clientMock.On("Do", mock.Anything).Return(&http.Response{
+ StatusCode: http.StatusCreated,
+ }, nil)
+
+ restclient.Client = &clientMock
+
+ producer := ProducerRegistrationInfo{
+ InfoProducerSupervisionCallbackUrl: "supervisionCallbackUrl",
+ SupportedInfoTypes: []string{"type1"},
+ InfoJobCallbackUrl: "jobCallbackUrl",
+ }
+
+ r := NewRegistratorImpl("http://localhost:9990")
+ err := r.RegisterProducer("Producer1", &producer)
+
+ assertions.Nil(err)
+ var actualRequest *http.Request
+ clientMock.AssertCalled(t, "Do", mock.MatchedBy(func(req *http.Request) bool {
+ actualRequest = req
+ return true
+ }))
+ assertions.Equal(http.MethodPut, actualRequest.Method)
+ assertions.Equal("http", actualRequest.URL.Scheme)
+ assertions.Equal("localhost:9990", actualRequest.URL.Host)
+ assertions.Equal("/data-producer/v1/info-producers/Producer1", actualRequest.URL.Path)
+ assertions.Equal("application/json; charset=utf-8", actualRequest.Header.Get("Content-Type"))
+ body, _ := ioutil.ReadAll(actualRequest.Body)
+ expectedBody := []byte(`{"info_producer_supervision_callback_url":"supervisionCallbackUrl","supported_info_types":["type1"],"info_job_callback_url":"jobCallbackUrl"}`)
+ assertions.Equal(expectedBody, body)
+ clientMock.AssertNumberOfCalls(t, "Do", 1)
+}
)
type Type struct {
- Name string
+ TypeId string
Schema string
}
var typeDir = "configs"
+var supportedTypes = make([]string, 0)
func GetTypes() ([]*Type, error) {
types := make([]*Type, 0, 1)
return types, nil
}
+func GetSupportedTypes() []string {
+ return supportedTypes
+}
+
func getType(path string) (*Type, error) {
fileName := filepath.Base(path)
typeName := strings.TrimSuffix(fileName, filepath.Ext(fileName))
if typeSchema, err := os.ReadFile(path); err == nil {
- return &Type{
- Name: typeName,
+ typeInfo := Type{
+ TypeId: typeName,
Schema: string(typeSchema),
- }, nil
+ }
+ supportedTypes = append(supportedTypes, typeName)
+ return &typeInfo, nil
} else {
return nil, err
}
const type1Schema = `{"title": "Type 1"}`
-func TestGetTypes_filesOkShouldReturnSliceOfTypes(t *testing.T) {
+func TestGetTypes_filesOkShouldReturnSliceOfTypesAndProvideSupportedTypes(t *testing.T) {
assertions := require.New(t)
typesDir, err := os.MkdirTemp("", "configs")
if err != nil {
}
types, err := GetTypes()
wantedType := Type{
- Name: "type1",
+ TypeId: "type1",
Schema: type1Schema,
}
wantedTypes := []*Type{&wantedType}
assertions.EqualValues(wantedTypes, types)
assertions.Nil(err)
+
+ supportedTypes := GetSupportedTypes()
+ assertions.EqualValues([]string{"type1"}, supportedTypes)
}
}
log.Debug("Initializing DMaaP Mediator Producer")
- if configuration.JobResultUri == "" {
- log.Fatal("Missing JOB_RESULT_URI")
+ if configuration.InfoJobCallbackUrl == "" {
+ log.Fatal("Missing INFO_JOB_CALLBACK_URL")
+ }
+ if configuration.InfoProducerSupervisionCallbackUrl == "" {
+ log.Fatal("Missing INFO_PRODUCER_SUPERVISION_CALLBACK_URL")
}
registrator := config.NewRegistratorImpl(configuration.InfoCoordinatorAddress)
} else {
log.Fatalf("Unable to get types to register due to: %v", err)
}
+ producer := config.ProducerRegistrationInfo{
+ InfoProducerSupervisionCallbackUrl: configuration.InfoProducerSupervisionCallbackUrl,
+ SupportedInfoTypes: jobtypes.GetSupportedTypes(),
+ InfoJobCallbackUrl: configuration.InfoJobCallbackUrl,
+ }
+ if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil {
+ log.Fatalf("Unable to register producer due to: %v", err)
+ }
}
func main() {
Code Review / nonrtric.git / commitdiff