SPQRNodeClient { public MicroPipelineInstantiationResponse updatePipeline(final MicroPipelineConfiguration pipelineConfiguration) throws IOException, RemoteClientConnectionFailedException { MicroPipelineValidationResult cfgValidationResult = this.pipelineConfigurationValidator.validate(pipelineConfiguration); if(cfgValidationResult != MicroPipelineValidationResult.OK) return new MicroPipelineInstantiationResponse("", cfgValidationResult, "Failed to validate pipeline configuration"); if(StringUtils.isBlank(pipelineConfiguration.getId())) return new MicroPipelineInstantiationResponse("", MicroPipelineValidationResult.MISSING_PIPELINE_ID, "Failed to generated unique pipeline identifier"); StringBuffer url = new StringBuffer(this.processingNodeServiceBaseUrl).append("/pipelines/").append(pipelineConfiguration.getId()); if(logger.isDebugEnabled()) logger.debug("Updating or instantiating pipeline [id="+pipelineConfiguration.getId()+"] on processing node " + url.toString()); try { final WebTarget webTarget = this.restClient.target(url.toString()); return webTarget.request(MediaType.APPLICATION_JSON).accept(MediaType.APPLICATION_JSON).put(Entity.entity(pipelineConfiguration, MediaType.APPLICATION_JSON), MicroPipelineInstantiationResponse.class); } catch(Exception e) { throw new RemoteClientConnectionFailedException("Failed to establish a connection with the remote resource manager [url="+url.toString()+"]. Error: " + e.getMessage()); } } SPQRNodeClient(final String protocol, final String remoteHost, final int servicePort, final int adminPort, final Client client); MicroPipelineInstantiationResponse instantiatePipeline(final MicroPipelineConfiguration pipelineConfiguration); MicroPipelineInstantiationResponse updatePipeline(final MicroPipelineConfiguration pipelineConfiguration); MicroPipelineShutdownResponse shutdown(final String pipelineId); void shutdown();  }
@Test public void testUpdatePipeline_withValidInput() throws Exception { String processingNodeHost = "processing-node-localhost"; int processingNodeServicePort = 7070; int processingNodeAdminPort = 7071; MicroPipelineConfiguration sampleConfiguration = new MicroPipelineConfiguration(); sampleConfiguration.setId("Test"); MicroPipelineComponentConfiguration componentCfg = new MicroPipelineComponentConfiguration(); componentCfg.setId("test-component"); componentCfg.setName("test"); componentCfg.setSettings(null); componentCfg.setToQueue("test-destination-queue"); componentCfg.setType(MicroPipelineComponentType.SOURCE); componentCfg.setVersion("test-version"); sampleConfiguration.getComponents().add(componentCfg); sampleConfiguration.getQueues().add(new StreamingMessageQueueConfiguration("test-destination-queue")); MicroPipelineInstantiationResponse pipelineInstantiationResponse = new MicroPipelineInstantiationResponse(sampleConfiguration.getId(), MicroPipelineValidationResult.OK, ""); String url = new StringBuffer("http: Client clientMock = Mockito.mock(Client.class); WebTarget webTargetMock = Mockito.mock(WebTarget.class); Builder b1Mock = Mockito.mock(Builder.class); Builder b2Mock = Mockito.mock(Builder.class); Mockito.when(clientMock.target(url)).thenReturn(webTargetMock); Mockito.when(webTargetMock.request(MediaType.APPLICATION_JSON)).thenReturn(b1Mock); Mockito.when(b1Mock.accept(MediaType.APPLICATION_JSON)).thenReturn(b2Mock); Mockito.when(b2Mock.put(Mockito.isA(Entity.class), Mockito.eq(MicroPipelineInstantiationResponse.class))).thenReturn(pipelineInstantiationResponse); SPQRNodeClient client = new SPQRNodeClient("http", processingNodeHost, processingNodeServicePort, processingNodeAdminPort, clientMock); MicroPipelineInstantiationResponse response = client.updatePipeline(sampleConfiguration); Assert.assertNotNull("Response must not be null", response); Assert.assertEquals("State must be 'OK'", MicroPipelineValidationResult.OK, response.getState()); Mockito.verify(clientMock).target(url); Mockito.verify(webTargetMock).request(MediaType.APPLICATION_JSON); Mockito.verify(b1Mock).accept(MediaType.APPLICATION_JSON); Mockito.verify(b2Mock).put(Mockito.isA(Entity.class), Mockito.eq(MicroPipelineInstantiationResponse.class)); }