class ImageClassificationWorkItemHandler extends AbstractLogOrThrowWorkItemHandler implements Cacheable {

	public void executeWorkItem(WorkItem workItem, WorkItemManager manager) {
		try {
			InputStream imageIS = null;
			Object inputImage = workItem.getParameter(PARAM_IMAGE);
			if (inputImage instanceof InputStream) {
				imageIS = (InputStream) inputImage;
			} else if (inputImage instanceof File) {
				imageIS = new FileInputStream((File) inputImage);
			} else if (inputImage instanceof URL) {
				imageIS = ((URL) inputImage).openStream();
			} else {
				throw new IllegalArgumentException("Input Parameter " + PARAM_IMAGE + " has an invalid type.");
			}
			Map<String, Object> results = new HashMap<String, Object>();
			Map<String, Double> predictions = classifier.classifyImage(imageIS);
			results.put(RESULT_PREDICTIONS, predictions);
			results.put(RESULT_PREDICTION, getPrediction(predictions));
			manager.completeWorkItem(workItem.getId(), results);
		} catch (Throwable cause) {
			System.out.println("Error running WIH: " + cause.getMessage());
			handleException(cause);
		}
	}

}

class ImageClassificationWorkItemHandlerTest {

	@Test
	public void testHandler() throws Exception {

		WorkItemImpl workItem = new WorkItemImpl();
		TestWorkItemManager manager = new TestWorkItemManager();
		ImageClassificationWorkItemHandler handler = new ImageClassificationWorkItemHandler();
		InputStream testImageIS = ImageClassificationWorkItemHandlerTest.class.getResourceAsStream("/my_bike.jpg");
		workItem.setParameter(ImageClassificationWorkItemHandler.PARAM_IMAGE, testImageIS);
		handler.executeWorkItem(workItem, manager);
		assertNotNull(manager.getResults());
		assertEquals(1, manager.getResults().size());
		assertEquals("mountain_bike", manager.getResults().get(workItem.getId()).get("prediction"));
		assertTrue(manager.getResults().containsKey(workItem.getId()));
	}

}