DeploymentUnitResource { @POST @Path("/load") @Produces(MediaType.APPLICATION_JSON) public Response loadDeploymentUnit(@QueryParam("name") String name, @QueryParam("version") Integer version, @QueryParam("replace") @DefaultValue("false") Boolean replaceOld) { if(name == null || name.length() == 0 || version == null || version < 0) { throw new WebApplicationException(buildResponse(Response.Status.BAD_REQUEST, "deploymentUnit name or version is invalid")); } DeploymentUnit loadedUnit; try { loadedUnit = deploymentUnitManager.load(name, version); logger.info("Successfully loaded deploymentUnit: " + loadedUnit.getName() + "/" + loadedUnit.getVersion()); } catch (DuplicateDeploymentUnitException e) { logger.error("Received request to load Deployment unit with name: " + name + " and version: " + version + ", which is already loaded. Discarding it."); return buildResponse(Response.Status.CONFLICT, e.getMessage()); } catch (Exception e) { String errMsg = "Could not load deploymentUnit: " + name + "/" + version; logger.error(errMsg, e); return buildResponse(Response.Status.INTERNAL_SERVER_ERROR, errMsg, e); } loadedUnit.getTaskMethods().values().stream().forEach(m -> { String routerName = new MethodId(m).getPrefix(); routerRegistry.createOrResize(routerName, taskRouterUtil.getConcurrency(loadedUnit, routerName)); }); if(replaceOld) { logger.info("Unloading redundant deploymentUnits"); for(DeploymentUnit unit: getRedundantUnits(loadedUnit)) { logger.info("Unloading deploymentUnit: " + unit.getName() + "/" + unit.getVersion()); deploymentUnitManager.unload(unit.getName(), unit.getVersion()); } } return buildResponse(Response.Status.OK, "Successfully loaded deploymentUnit: " + name + "/" + loadedUnit.getVersion()); } @Inject  DeploymentUnitResource(DeploymentUnitsManager deploymentUnitsManager, RouterRegistry routerRegistry,
                                  TaskRouterUtil taskRouterUtil); @POST @Path("/load") @Produces(MediaType.APPLICATION_JSON) Response loadDeploymentUnit(@QueryParam("name") String name, @QueryParam("version") Integer version,
                                       @QueryParam("replace") @DefaultValue("false") Boolean replaceOld); @POST @Path("/unload") @Produces(MediaType.APPLICATION_JSON) Response unloadDeploymentUnit(@QueryParam("name") String name, @QueryParam("version") Integer version);  }
@Test public void testLoadDeploymentUnit() throws Exception { DeploymentUnit unit1 = getMockedDUnit("unit1", 1, "task1", getMethod1()); DeploymentUnit unit2 = getMockedDUnit("unit2", 1, "task2", getMethod2()); DeploymentUnit unit1_new = getMockedDUnit("unit1", 2, "task1", getMethod1()); when(deploymentUnitsManager.getAllDeploymentUnits()).thenReturn(Arrays.asList(unit1, unit2)); when(deploymentUnitsManager.load("unit1", 2)).thenReturn(unit1_new); Response response = deploymentUnitResource.loadDeploymentUnit("unit1", 2, false); Assertions.assertThat(response.getStatus()).isEqualTo(200); verify(deploymentUnitsManager).load("unit1", 2); verify(routerRegistry).createOrResize("com.flipkart.flux.resource.DeploymentUnitResourceTest_testMethod", defaultTaskConcurrency); verifyNoMoreInteractions(deploymentUnitsManager); verifyNoMoreInteractions(routerRegistry); }
@Test public void testLoadDeploymentUnitReplace_shouldUnloadPreviousDeploymentUnit() throws Exception { DeploymentUnit unit1 = getMockedDUnit("unit1", 1, "task1", getMethod1()); DeploymentUnit unit2 = getMockedDUnit("unit2", 1, "task2", getMethod2()); DeploymentUnit unit1_new = getMockedDUnit("unit1", 2, "task1", getMethod1()); when(deploymentUnitsManager.getAllDeploymentUnits()).thenReturn(Arrays.asList(unit1, unit2)); when(deploymentUnitsManager.load("unit1", 2)).thenReturn(unit1_new); Response response = deploymentUnitResource.loadDeploymentUnit("unit1", 2, true); Assertions.assertThat(response.getStatus()).isEqualTo(200); verify(deploymentUnitsManager).load("unit1", 2); verify(routerRegistry).createOrResize("com.flipkart.flux.resource.DeploymentUnitResourceTest_testMethod", 3); verify(deploymentUnitsManager).getAllDeploymentUnits(); verify(deploymentUnitsManager).unload("unit1", 1); verifyNoMoreInteractions(deploymentUnitsManager); verifyNoMoreInteractions(routerRegistry); }