At Job create, error will be returned if no producers accepted the job.
Change-Id: I63422a4d898f21510667d7439fc3fddb6d3a5170
Issue-ID: NONRTRIC-173
Signed-off-by: PatrikBuhr <patrik.buhr@est.tech>
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
/**
* Callbacks to the EiProducer
*/
@Autowired
ApplicationConfig applicationConfig;
- public void notifyProducersJobCreated(EiJob eiJob) {
- for (EiProducer producer : eiJob.type().getProducers()) {
- notifyProducerJobStarted(producer, eiJob);
- }
- }
-
public void notifyProducersJobDeleted(EiJob eiJob) {
AsyncRestClient restClient = restClient(false);
ProducerJobInfo request = new ProducerJobInfo(eiJob);
}
}
- public void notifyProducerJobStarted(EiProducer producer, EiJob eiJob) {
+ public Mono<Integer> notifyProducersJobStarted(EiJob eiJob) {
+ return Flux.fromIterable(eiJob.type().getProducers()) //
+ .flatMap(eiProducer -> notifyProducerJobStarted(eiProducer, eiJob)) //
+ .collectList() //
+ .flatMap(okResponses -> Mono.just(Integer.valueOf(okResponses.size()))); //
+ }
+
+ public Mono<String> notifyProducerJobStarted(EiProducer producer, EiJob eiJob) {
AsyncRestClient restClient = restClient(false);
ProducerJobInfo request = new ProducerJobInfo(eiJob);
String body = gson.toJson(request);
- restClient.post(producer.jobCreationCallbackUrl(), body) //
- .subscribe(notUsed -> logger.debug("Job subscription started OK {}", producer.id()), //
- throwable -> logger.warn("Job subscription failed {}", producer.id(), throwable.toString()), null);
-
+ return restClient.post(producer.jobCreationCallbackUrl(), body)
+ .doOnNext(resp -> logger.debug("Job subscription started OK {}", producer.id()))
+ .onErrorResume(throwable -> {
+ logger.warn("Job subscription failed {}", producer.id(), throwable.toString());
+ return Mono.empty();
+ });
}
private AsyncRestClient restClient(boolean useTrustValidation) {
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
+import org.oransc.enrichment.exceptions.ServiceException;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
this.message = message;
}
- public static Mono<ResponseEntity<Object>> createMono(Exception e, HttpStatus code) {
+ public static Mono<ResponseEntity<Object>> createMono(Throwable e, HttpStatus code) {
return Mono.just(create(e, code));
}
- public static ResponseEntity<Object> create(Exception e, HttpStatus code) {
+ public static ResponseEntity<Object> create(Throwable e, HttpStatus code) {
if (e instanceof RuntimeException) {
code = HttpStatus.INTERNAL_SERVER_ERROR;
+ } else if (e instanceof ServiceException) {
+ ServiceException se = (ServiceException) e;
+ if (se.getHttpStatus() != null) {
+ code = se.getHttpStatus();
+ }
}
return create(e.toString(), code);
}
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
@SuppressWarnings("java:S3457") // No need to call "toString()" method as formatting and string ..
@RestController("ConsumerController")
code = 404,
message = "Enrichment Information type is not found",
response = ErrorResponse.ErrorInfo.class)})
- public ResponseEntity<Object> putIndividualEiJob( //
+ public Mono<ResponseEntity<Object>> putIndividualEiJob( //
@PathVariable("eiTypeId") String eiTypeId, //
@PathVariable("eiJobId") String eiJobId, //
@RequestBody ConsumerEiJobInfo eiJobInfo) {
+
+ final boolean isNewJob = this.eiJobs.get(eiJobId) == null;
+
+ return validatePutEiJob(eiTypeId, eiJobId, eiJobInfo) //
+ .flatMap(this::notifyProducersNewJob) //
+ .doOnNext(newEiJob -> this.eiJobs.put(newEiJob)) //
+ .flatMap(newEiJob -> Mono.just(new ResponseEntity<>(isNewJob ? HttpStatus.CREATED : HttpStatus.OK)))
+ .onErrorResume(throwable -> Mono.just(ErrorResponse.create(throwable, HttpStatus.NOT_FOUND)));
+ }
+
+ private Mono<EiJob> notifyProducersNewJob(EiJob newEiJob) {
+ return this.producerCallbacks.notifyProducersJobStarted(newEiJob) //
+ .flatMap(noOfAcceptingProducers -> {
+ if (noOfAcceptingProducers.intValue() > 0) {
+ return Mono.just(newEiJob);
+ } else {
+ return Mono.error(new ServiceException("Job not accepted by any producers", HttpStatus.CONFLICT));
+ }
+ });
+ }
+
+ private Mono<EiJob> validatePutEiJob(String eiTypeId, String eiJobId, ConsumerEiJobInfo eiJobInfo) {
try {
EiType eiType = this.eiTypes.getType(eiTypeId);
- validateJobData(eiType.getJobDataSchema(), eiJobInfo.jobData);
+ validateJsonObjectAgainstSchema(eiType.getJobDataSchema(), eiJobInfo.jobData);
EiJob existingEiJob = this.eiJobs.get(eiJobId);
- final boolean newJob = existingEiJob == null;
+
if (existingEiJob != null && !existingEiJob.type().getId().equals(eiTypeId)) {
- return ErrorResponse.create("Not allowed to change type for existing EI job", HttpStatus.CONFLICT);
+ throw new ServiceException("Not allowed to change type for existing EI job", HttpStatus.CONFLICT);
}
- EiJob newEiJob = toEiJob(eiJobInfo, eiJobId, eiType);
- this.eiJobs.put(newEiJob);
- this.producerCallbacks.notifyProducersJobCreated(newEiJob);
- return new ResponseEntity<>(newJob ? HttpStatus.CREATED : HttpStatus.OK);
+ return Mono.just(toEiJob(eiJobInfo, eiJobId, eiType));
} catch (Exception e) {
- return ErrorResponse.create(e, HttpStatus.NOT_FOUND);
+ return Mono.error(e);
}
}
- private void validateJobData(Object schemaObj, Object object) throws ServiceException {
- if (schemaObj == null) {
- return; // schema is optional for now
- }
- try {
- ObjectMapper mapper = new ObjectMapper();
+ private void validateJsonObjectAgainstSchema(Object schemaObj, Object object) throws ServiceException {
+ if (schemaObj != null) { // schema is optional for now
+ try {
+ ObjectMapper mapper = new ObjectMapper();
- String schemaAsString = mapper.writeValueAsString(schemaObj);
- JSONObject schemaJSON = new JSONObject(schemaAsString);
- Schema schema = SchemaLoader.load(schemaJSON);
+ String schemaAsString = mapper.writeValueAsString(schemaObj);
+ JSONObject schemaJSON = new JSONObject(schemaAsString);
+ Schema schema = SchemaLoader.load(schemaJSON);
- String objectAsString = mapper.writeValueAsString(object);
- JSONObject json = new JSONObject(objectAsString);
- schema.validate(json);
- } catch (Exception e) {
- throw new ServiceException("Json validation failure", e);
+ String objectAsString = mapper.writeValueAsString(object);
+ JSONObject json = new JSONObject(objectAsString);
+ schema.validate(json);
+ } catch (Exception e) {
+ throw new ServiceException("Json validation failure " + e.toString(), HttpStatus.CONFLICT);
+ }
}
-
}
// Status TBD
import org.immutables.gson.Gson;
@Gson.TypeAdapters
-@ApiModel(value = "ei_job_info", description = "Information for a Enrichment Information Job")
+@ApiModel(value = "EiJob", description = "Information for an Enrichment Information Job")
public class ConsumerEiJobInfo {
@ApiModelProperty(value = "Identity of the owner of the job", required = true)
- @SerializedName("job_owner")
- @JsonProperty(value = "job_owner", required = true)
+ @SerializedName("jobOwner")
+ @JsonProperty(value = "jobOwner", required = true)
public String owner;
@ApiModelProperty(value = "EI Type specific job data", required = true)
- @SerializedName("job_data")
- @JsonProperty(value = "job_data", required = true)
+ @SerializedName("jobParameters")
+ @JsonProperty(value = "jobParameters", required = true)
public Object jobData;
@ApiModelProperty(value = "The target of the EI data", required = true)
- @SerializedName("target_uri")
- @JsonProperty(value = "target_uri", required = true)
+ @SerializedName("targetUri")
+ @JsonProperty(value = "targetUri", required = true)
public String targetUri;
public ConsumerEiJobInfo() {
import org.immutables.gson.Gson;
@Gson.TypeAdapters
-@ApiModel(value = "ei_job_status", description = "Status for an EI Job")
+@ApiModel(value = "EiJobStatus", description = "Status for an EI Job")
public class ConsumerEiJobStatus {
@Gson.TypeAdapters
- @ApiModel(value = "operational_state", description = "Represents the operational states for a EI Job")
+ @ApiModel(value = "OperationalState", description = "Represents the operational states for a EI Job")
public enum OperationalState {
ENABLED, DISABLED
}
+ "DISABLED: TBD.";
@ApiModelProperty(value = OPERATIONAL_STATE_DESCRIPTION, name = "operational_state", required = true)
- @SerializedName("operational_state")
- @JsonProperty(value = "operational_state", required = true)
+ @SerializedName("operationalState")
+ @JsonProperty(value = "operationalState", required = true)
public final OperationalState state;
public ConsumerEiJobStatus(OperationalState state) {
import org.immutables.gson.Gson;
@Gson.TypeAdapters
-@ApiModel(value = "ei_type_info", description = "Information for an EI type")
+@ApiModel(value = "EiType", description = "Information for an EI type")
public class ConsumerEiTypeInfo {
@ApiModelProperty(value = "Json schema for the job data")
- @SerializedName("job_data_schema")
- @JsonProperty("job_data_schema")
- public final Object jobDataSchema;
+ @SerializedName("eiJobParametersSchema")
+ @JsonProperty("eiJobParametersSchema")
+ public final Object jobParametersSchema;
- ConsumerEiTypeInfo(Object jobDataSchema) {
- this.jobDataSchema = jobDataSchema;
+ ConsumerEiTypeInfo(Object jobParametersSchema) {
+ this.jobParametersSchema = jobParametersSchema;
}
}
for (EiType type : types) {
for (EiJob job : this.eiJobs.getJobsForType(type)) {
- this.producerCallbacks.notifyProducerJobStarted(producer, job);
+ this.producerCallbacks.notifyProducerJobStarted(producer, job) //
+ .subscribe(//
+ response -> logger.debug("Producer notified OK"), //
+ throwable -> logger.warn("Producer rejected job {}", throwable.getMessage()) //
+ );
}
type.addProducer(producer);
}
package org.oransc.enrichment.exceptions;
+import lombok.Getter;
+
+import org.springframework.http.HttpStatus;
+
public class ServiceException extends Exception {
private static final long serialVersionUID = 1L;
+ @Getter
+ private final HttpStatus httpStatus;
public ServiceException(String message) {
super(message);
+ this.httpStatus = null;
}
public ServiceException(String message, Exception originalException) {
super(message, originalException);
+ this.httpStatus = null;
+ }
+
+ public ServiceException(String message, HttpStatus httpStatus) {
+ super(message);
+ this.httpStatus = httpStatus;
}
}
import org.oransc.enrichment.controller.ProducerSimulatorController;
import org.oransc.enrichment.controllers.consumer.ConsumerConsts;
import org.oransc.enrichment.controllers.consumer.ConsumerEiJobInfo;
+import org.oransc.enrichment.controllers.consumer.ConsumerEiTypeInfo;
import org.oransc.enrichment.controllers.producer.ProducerConsts;
import org.oransc.enrichment.controllers.producer.ProducerRegistrationInfo;
import org.oransc.enrichment.controllers.producer.ProducerRegistrationInfo.ProducerEiTypeRegistrationInfo;
putEiProducerWithOneType(EI_PRODUCER_ID, "test");
String url = ConsumerConsts.API_ROOT + "/eitypes/test";
String rsp = restClient().get(url).block();
- assertThat(rsp).contains("job_data_schema");
+ ConsumerEiTypeInfo info = gson.fromJson(rsp, ConsumerEiTypeInfo.class);
+ assertThat(info.jobParametersSchema).isNotNull();
}
@Test
putEiJob(EI_TYPE_ID, "jobId");
String url = ConsumerConsts.API_ROOT + "/eitypes/typeId/eijobs/jobId";
String rsp = restClient().get(url).block();
- assertThat(rsp).contains("job_data");
+ ConsumerEiJobInfo info = gson.fromJson(rsp, ConsumerEiJobInfo.class);
+ assertThat(info.owner).isEqualTo("owner");
}
@Test
@Test
void testPutEiJob() throws Exception {
+ // Test that one producer accepting a job is enough
putEiProducerWithOneType(EI_PRODUCER_ID, EI_TYPE_ID);
+ putEiProducerWithOneTypeRejecting("simulateProducerError", EI_TYPE_ID);
String url = ConsumerConsts.API_ROOT + "/eitypes/typeId/eijobs/jobId";
String body = gson.toJson(eiJobInfo());
ProducerJobInfo request = simulatorResults.jobsStarted.get(0);
assertThat(request.id).isEqualTo("jobId");
+ assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(1);
+
resp = restClient().putForEntity(url, body).block();
assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
EiJob job = this.eiJobs.getJob("jobId");
assertThat(job.owner()).isEqualTo("owner");
}
+ @Test
+ void putEiProducerWithOneType_rejecting() throws JsonMappingException, JsonProcessingException, ServiceException {
+ putEiProducerWithOneTypeRejecting("simulateProducerError", EI_TYPE_ID);
+ String url = ConsumerConsts.API_ROOT + "/eitypes/typeId/eijobs/jobId";
+ String body = gson.toJson(eiJobInfo());
+ testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "Job not accepted by any producers");
+
+ ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
+ assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(1);
+ }
+
@Test
void testPutEiJob_jsonSchemavalidationError() throws Exception {
putEiProducerWithOneType(EI_PRODUCER_ID, EI_TYPE_ID);
new ConsumerEiJobInfo(jsonObject("{ \"XXstring\" : \"value\" }"), "owner", "targetUri");
String body = gson.toJson(jobInfo);
- testErrorCode(restClient().put(url, body), HttpStatus.NOT_FOUND, "Json validation failure");
+ testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "Json validation failure");
}
@Test
return new ProducerEiTypeRegistrationInfo(jsonSchemaObject(), typeId);
}
+ ProducerRegistrationInfo producerEiRegistratioInfoRejecting(String typeId)
+ throws JsonMappingException, JsonProcessingException {
+ Collection<ProducerEiTypeRegistrationInfo> types = new ArrayList<>();
+ types.add(producerEiTypeRegistrationInfo(typeId));
+ return new ProducerRegistrationInfo(types, baseUrl() + ProducerSimulatorController.JOB_CREATED_ERROR_URL,
+ baseUrl() + ProducerSimulatorController.JOB_DELETED_ERROR_URL);
+ }
+
ProducerRegistrationInfo producerEiRegistratioInfo(String typeId)
throws JsonMappingException, JsonProcessingException {
Collection<ProducerEiTypeRegistrationInfo> types = new ArrayList<>();
return this.eiJobs.getJob(jobId);
}
+ private EiType putEiProducerWithOneTypeRejecting(String producerId, String eiTypeId)
+ throws JsonMappingException, JsonProcessingException, ServiceException {
+ String url = ProducerConsts.API_ROOT + "/eiproducers/" + producerId;
+ String body = gson.toJson(producerEiRegistratioInfoRejecting(eiTypeId));
+
+ restClient().putForEntity(url, body).block();
+ return this.eiTypes.getType(eiTypeId);
+ }
+
private EiType putEiProducerWithOneType(String producerId, String eiTypeId)
throws JsonMappingException, JsonProcessingException, ServiceException {
String url = ProducerConsts.API_ROOT + "/eiproducers/" + producerId;
public static final String JOB_CREATED_URL = "/producer_simulator/job_created";
public static final String JOB_DELETED_URL = "/producer_simulator/job_deleted";
+ public static final String JOB_CREATED_ERROR_URL = "/producer_simulator/job_created_error";
+ public static final String JOB_DELETED_ERROR_URL = "/producer_simulator/job_deleted_error";
public static class TestResults {
public List<ProducerJobInfo> jobsStarted = Collections.synchronizedList(new ArrayList<ProducerJobInfo>());
public List<ProducerJobInfo> jobsStopped = Collections.synchronizedList(new ArrayList<ProducerJobInfo>());
+ public int noOfRejectedCreate = 0;
+ public int noOfRejectedDelete = 0;
public boolean errorFound = false;
public TestResults() {
jobsStarted.clear();
jobsStopped.clear();
this.errorFound = false;
+ this.noOfRejectedCreate = 0;
+ this.noOfRejectedDelete = 0;
}
}
}
}
+ @PostMapping(path = JOB_CREATED_ERROR_URL, produces = MediaType.APPLICATION_JSON_VALUE)
+ @ApiOperation(value = "Callback for EI job creation, returns error", notes = "")
+ @ApiResponses(
+ value = { //
+ @ApiResponse(code = 200, message = "OK", response = void.class)}//
+ )
+ public ResponseEntity<Object> jobCreatedCallbackReturnError( //
+ @RequestBody ProducerJobInfo request) {
+ logger.info("Job created (returning error) callback {}", request.id);
+ this.testResults.noOfRejectedCreate += 1;
+ return ErrorResponse.create("Producer returns error on create job", HttpStatus.NOT_FOUND);
+ }
+
+ @PostMapping(path = JOB_DELETED_ERROR_URL, produces = MediaType.APPLICATION_JSON_VALUE)
+ @ApiOperation(value = "Callback for EI job creation, returns error", notes = "")
+ @ApiResponses(
+ value = { //
+ @ApiResponse(code = 200, message = "OK", response = void.class)}//
+ )
+ public ResponseEntity<Object> jobDeletedCallbackReturnError( //
+ @RequestBody ProducerJobInfo request) {
+ logger.info("Job created (returning error) callback {}", request.id);
+ this.testResults.noOfRejectedDelete += 1;
+ return ErrorResponse.create("Producer returns error on delete job", HttpStatus.NOT_FOUND);
+ }
+
}
Code Review / nonrtric.git / commitdiff