[Task]: Ts/1024 calibrate tool read from mq

packages/client/hmi-client/src/components/workflow/tera-calibration-node-ciemss.vue

@@ -125,6 +125,8 @@ import { ChartConfig, RunResults } from '@/types/SimulateConfig';
 import { workflowEventBus } from '@/services/workflow';
 import _ from 'lodash';
 import { Poller, PollerState } from '@/api/api';
+import useAuthStore from '@/stores/auth';
+import { EventSourcePolyfill } from 'event-source-polyfill';
 import {
 	CalibrationOperationCiemss,
 	CalibrationOperationStateCiemss,
@@ -237,20 +239,49 @@ const runCalibrate = async () => {
 	};
 	const response = await makeCalibrateJobCiemss(calibrationRequest);
 
+	// Run MQ Test
+	console.log(`Checking MQ for job Id ${response.simulationId}`);
+	const auth = useAuthStore();
+	const eventsource = new EventSourcePolyfill(
+		`/api/simulations/${response.simulationId}/partial-result`,
+		{
+			headers: {
+				Authorization: `Bearer ${auth.token}`
+			}
+		}
+	);
+	eventsource.onerror = (e) => {
+		console.log('An error occurred while attempting to connect.');
+		console.log(e);
+		eventsource.close();
+		console.log('Closing eventsource');
+	};
+	eventsource.onmessage = (event) => {
+		console.log(event);
+		const json = event.data;
+		console.log(json);
+		eventsource.close();
+		console.log('Closing eventsource');
+	};
 	if (response?.simulationId) {
 		getStatus(response.simulationId);
 	}
 };
 
 const getStatus = async (simulationId: string) => {
+	console.log('Getting status');
 	showSpinner.value = true;
-	if (!simulationId) return;
-
+	if (!simulationId) {
+		console.log('No sim id');
+		return;
+	}
+	console.log(`Simulation Id:${simulationId}`);
 	const runIds = [simulationId];
 	poller
 		.setInterval(3000)
 		.setThreshold(300)
 		.setPollAction(async () => simulationPollAction(runIds, props.node, progress, emit));
+	console.log('Poller defined');
 	const pollerResults = await poller.start();
 
 	if (pollerResults.state !== PollerState.Done || !pollerResults.data) {

packages/services/hmi-server/src/main/java/software/uncharted/terarium/hmiserver/models/SimulationIntermediateResults.java

@@ -11,5 +11,10 @@
 public class SimulationIntermediateResults implements Serializable {
 	@JsonAlias("job_id")
 	private String jobId;
-	private float progress;
+	private Double progress;
+
+	@Override
+	public String toString(){
+		return "{job_id: '" + this.jobId + "', progress: " + Double.toString(this.progress) + "}";
+	}
 }

packages/services/hmi-server/src/main/java/software/uncharted/terarium/hmiserver/resources/dataservice/SimulationResource.java

@@ -11,6 +11,7 @@
 import io.smallrye.mutiny.Multi;
 import org.eclipse.microprofile.rest.client.inject.RestClient;
 import org.eclipse.microprofile.reactive.messaging.Channel;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
 import org.reactivestreams.Publisher;
 import org.jboss.resteasy.annotations.SseElementType;
 import software.uncharted.terarium.hmiserver.models.dataservice.Assets;
@@ -23,6 +24,7 @@
 import software.uncharted.terarium.hmiserver.resources.SnakeCaseResource;
 import software.uncharted.terarium.hmiserver.utils.Converter;
 import software.uncharted.terarium.hmiserver.models.SimulationIntermediateResults;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.smallrye.reactive.messaging.annotations.Broadcast;
 import org.eclipse.microprofile.reactive.messaging.Emitter;
 import java.util.Map;
@@ -51,8 +53,9 @@ public class SimulationResource implements SnakeCaseResource {
 	@RestClient
 	DatasetProxy datasetProxy;
 
+	//TODO: https://github.com/DARPA-ASKEM/Terarium/issues/1757
 	@Inject
-	@Channel("simulationStatus") Publisher<SimulationIntermediateResults> partialSimulationResults;
+	@Channel("simulationStatus") Publisher<byte[]> partialSimulationStream;
 
 	@Broadcast
 	@Channel("simulationStatus")
@@ -162,25 +165,43 @@ public Response createFromSimulationResult(
 	@Produces(MediaType.SERVER_SENT_EVENTS)
 	@SseElementType(MediaType.APPLICATION_JSON)
 	@Tag(name = "Stream partial/intermediate simulation result associated with run ID")
-	public Publisher<SimulationIntermediateResults> stream(
+	public Publisher<byte[]> stream(
 		@PathParam("jobId") final String jobId
 	) {
-		return Multi.createFrom().publisher(partialSimulationResults).select().where(event -> event.getJobId().equals(jobId));
+		ObjectMapper mapper = new ObjectMapper();
+		return Multi.createFrom().publisher(partialSimulationStream).filter(event -> {
+			try{ 
+				//TODO: https://github.com/DARPA-ASKEM/Terarium/issues/1757
+				String jsonString = new String(event);
+				jsonString = jsonString.replace(" ","");
+
+				SimulationIntermediateResults interResult = mapper.readValue(jsonString, SimulationIntermediateResults.class);
+
+				return interResult.getJobId().equals(jobId);
+			}
+			catch(Exception e){
+				log.error("Error occured while trying to convert simulation-status message to type: SimulationIntermediateResults");
+				log.error(event.toString());
+				log.error(e.toString());
+				return false;
+			}
+		});
 	}
 
 	// When we finalize the SimulationIntermediateResults object this end point will need to be passed more parameters
+	//TODO: https://github.com/DARPA-ASKEM/Terarium/issues/1757
 	@PUT
 	@Path("/{jobId}/create-partial-result")
 	@Produces(MediaType.APPLICATION_JSON)
 	@Tag(name = "Used to write to the simulation status channel providing a job ID")
 	public Response createPartialResult(
 		@PathParam("jobId") final String jobId
 	) {
-		float progress = 0.1f;
-		final SimulationIntermediateResults event = new SimulationIntermediateResults();
+		Double progress = 0.01;
+		SimulationIntermediateResults event = new SimulationIntermediateResults();
 		event.setJobId(jobId);
 		event.setProgress(progress);
 		partialSimulationEmitter.send(event);
-		return Response.ok(Map.of("id", jobId.toString())).build();
+		return Response.ok().build();
 	}
 }