support fetching resources with same parent table #144

batch/src/main/java/org/openmrs/analytics/FhirEtl.java

@@ -175,19 +175,20 @@ static void runFhirJdbcFetch(FhirEtlOptions options, FhirContext fhirContext)
 		JdbcIO.DataSourceConfiguration jdbcConfig = jdbcUtil.getJdbcConfig();
 		int batchSize = Math.min(options.getBatchSize(), 170); // batch size > 200 will result in HTTP 400 Bad Request
 		int jdbcFetchSize = options.getJdbcFetchSize();
-		Map<String, String> reverseMap = jdbcUtil.createFhirReverseMap(options.getSearchList(),
+		Map<String, List<String>> reverseMap = jdbcUtil.createFhirReverseMap(options.getSearchList(),
 		    options.getTableFhirMapPath());
 		// process each table-resource mappings
-		for (Map.Entry<String, String> entry : reverseMap.entrySet()) {
+		for (Map.Entry<String, List<String>> entry : reverseMap.entrySet()) {
 			String tableName = entry.getKey();
-			String resourceType = entry.getValue();
-			String baseBundleUrl = options.getOpenmrsServerUrl() + options.getServerFhirEndpoint() + "/" + resourceType;
 			int maxId = jdbcUtil.fetchMaxId(tableName);
 			Map<Integer, Integer> IdRanges = jdbcUtil.createIdRanges(maxId, jdbcFetchSize);
-			PCollection<SearchSegmentDescriptor> inputSegments = pipeline.apply(Create.of(IdRanges))
-			        .apply(new JdbcFetchUtil.FetchUuids(tableName, jdbcConfig))
-			        .apply(new JdbcFetchUtil.CreateSearchSegments(resourceType, baseBundleUrl, batchSize));
-			fetchSegments(inputSegments, resourceType, options);
+			for (String resourceType : entry.getValue()) {
+				String baseBundleUrl = options.getOpenmrsServerUrl() + options.getServerFhirEndpoint() + "/" + resourceType;
+				PCollection<SearchSegmentDescriptor> inputSegments = pipeline.apply(Create.of(IdRanges))
+				        .apply(new JdbcFetchUtil.FetchUuids(tableName, jdbcConfig))
+				        .apply(new JdbcFetchUtil.CreateSearchSegments(resourceType, baseBundleUrl, batchSize));
+				fetchSegments(inputSegments, resourceType, options);
+			}
 		}
 		PipelineResult result = pipeline.run();
 		result.waitUntilFinish();

batch/src/main/java/org/openmrs/analytics/JdbcFetchUtil.java

@@ -167,21 +167,37 @@ public Map<Integer, Integer> createIdRanges(int count, int rangeSize) {
 		return rangeMap;
 	}
 
-	public Map<String, String> createFhirReverseMap(String searchString, String tableFhirMapPath) throws IOException {
+	/**
+	 * Creates a map from database table names to the list of FHIR resources that correspond to that
+	 * table. For example: person->Person,Patient and visit->Encounter.
+	 *
+	 * @param searchString the comma separated list of FHIR resources we care about.
+	 * @param tableFhirMapPath the file that contains the general configuration.
+	 * @return the computed map.
+	 */
+	public Map<String, List<String>> createFhirReverseMap(String searchString, String tableFhirMapPath) throws IOException {
 		Gson gson = new Gson();
 		Path pathToFile = Paths.get(tableFhirMapPath);
 		try (Reader reader = Files.newBufferedReader(pathToFile.toAbsolutePath(), StandardCharsets.UTF_8)) {
 			GeneralConfiguration generalConfiguration = gson.fromJson(reader, GeneralConfiguration.class);
 			Map<String, EventConfiguration> tableToFhirMap = generalConfiguration.getEventConfigurations();
 			String[] searchList = searchString.split(",");
-			Map<String, String> reverseMap = new HashMap<String, String>();
+			Map<String, List<String>> reverseMap = new HashMap<String, List<String>>();
 			for (Map.Entry<String, EventConfiguration> entry : tableToFhirMap.entrySet()) {
 				Map<String, String> linkTemplate = entry.getValue().getLinkTemplates();
 				for (String search : searchList) {
 					if (linkTemplate.containsKey("fhir") && linkTemplate.get("fhir") != null) {
 						String[] resourceName = linkTemplate.get("fhir").split("/");
 						if (resourceName.length >= 1 && resourceName[1].equals(search)) {
-							reverseMap.put(entry.getValue().getParentTable(), resourceName[1]);
+							if (reverseMap.containsKey(entry.getValue().getParentTable())) {
+								List<String> resources = reverseMap.get(entry.getValue().getParentTable());
+								resources.add(resourceName[1]);
+							} else {
+								List<String> resources = new ArrayList<String>();
+								resources.add(resourceName[1]);
+								reverseMap.put(entry.getValue().getParentTable(), resources);
+							}
+
 						}
 					}
 				}

batch/src/test/java/org/openmrs/analytics/JdbcFetchUtilTest.java

@@ -131,16 +131,16 @@ public void testCreateSearchSegmentDescriptor() {
 	}
 
 	@Test
-	public void testCreateFhirReverseMap() throws IOException {
-		// here we pass Encounters as such we expect visits to be included in the reverseMap as well
-		Map<String, String> reverseMap = jdbcFetchUtil.createFhirReverseMap("Patient,Encounter,Observation",
+	public void testCreateFhirReverseMap() throws Exception {
+		Map<String, List<String>> reverseMap = jdbcFetchUtil.createFhirReverseMap("Patient,Person,Encounter,Observation",
 		    "../utils/dbz_event_to_fhir_config.json");
-		// we expect 4 objects, and visit should be included
-		assertEquals(4, reverseMap.size());// not 3
-		assertFalse(reverseMap.get("visit").isEmpty());
-		assertFalse(reverseMap.get("encounter").isEmpty());
-		assertFalse(reverseMap.get("obs").isEmpty());
-		assertFalse(reverseMap.get("person").isEmpty());
+
+		assertEquals(reverseMap.size(), 4);
+		assertEquals(reverseMap.get("person").size(), 2);
+		assertTrue(reverseMap.get("person").contains("Patient"));
+		assertTrue(reverseMap.get("person").contains("Person"));
+		assertTrue(reverseMap.get("encounter").contains("Encounter"));
+		assertTrue(reverseMap.get("visit").contains("Encounter"));
+		assertTrue(reverseMap.get("obs").contains("Observation"));
 	}
-
 }

utils/dbz_event_to_fhir_config.json

@@ -243,4 +243,4 @@
       }
     }
   }
-}
+}