GetMessageColumns extends MetadataAPIBase { public Set<String> getColumnNames(Integer pid) { Process selectedProcess = processDAO.get(pid); Process parentProcess = selectedProcess.getProcess(); Integer parentProcessId = parentProcess.getProcessId(); String nextProcessOfParent = processDAO.get(parentProcessId).getNextProcessId(); listOfSourcesForGivenPid.clear(); prevMap.clear(); listOfSourcePids.clear(); for (String nextProcessId : nextProcessOfParent.split(",")) { listOfSourcePids.add(Integer.valueOf(nextProcessId)); } Map<Integer, String> nextPidMap = new HashMap<Integer, String>(); nextPidMap.put(parentProcessId, nextProcessOfParent); List<com.wipro.ats.bdre.md.dao.jpa.Process> jpaProcessList = processDAO.subProcesslist(parentProcessId); for (Process subProcess : jpaProcessList) { nextPidMap.put(subProcess.getProcessId(), subProcess.getNextProcessId()); } List<Integer> currentUpstreamList = new ArrayList(); currentUpstreamList.addAll(listOfSourcePids); for (Integer sourcePid : listOfSourcePids) { prevMap.put(sourcePid, null); } while (!currentUpstreamList.isEmpty()) { System.out.println("currentUpstreamList = " + currentUpstreamList); GetMessageColumns getMessageColumns = new GetMessageColumns(); System.out.println(" calling identifyflows"); getMessageColumns.identifyFlows(currentUpstreamList, nextPidMap,parentProcessId); } System.out.println("prevMap = " + prevMap); List<Integer> sourcesForGivenNode = new GetMessageColumns().recursivelyGetSources(pid); Set<String> columnNames = new LinkedHashSet<>(); for(Integer sourcePid : sourcesForGivenNode){ List<Properties> messageProperties = propertiesDAO.getPropertiesForConfig(sourcePid, "message"); if(messageProperties != null){ Properties messageProperty = messageProperties.get(0); String messageName = messageProperty.getPropValue(); Messages message = messagesDAO.get(messageName); String schema = message.getMessageSchema(); String[] columnAndDataTypes = schema.split(","); for(String s : columnAndDataTypes){ columnNames.add(s.split(":")[0]); } } } return columnNames; } GetMessageColumns(); static void main(String[] args); Set<String> getColumnNames(Integer pid); HashMap<Integer,String> getMessageList(Integer pid); Set<String> getMessageColumnNames(Integer pid); List<Integer> recursivelyGetNonEmptySchema(Integer pid); List<Integer> recursivelyGetSources(Integer pid); void identifyFlows(List<Integer> currentUpstreamList, Map<Integer,String> nextPidMap, Integer parentProcessId); void add(Integer key, Integer newValue, Map<Integer,Set<Integer>> prevMap); @Override Object execute(String[] params); static Map<Integer,Set<Integer>> prevMap; static List<Integer> listOfSourcePids; static List<Integer> listOfTransformations; static List<Integer> listOfEmitters; static List<Integer> listOfSourcesForGivenPid; static List<Integer> listOfPidWithNonEmptySchema; }
@Ignore @Test public void testGetParentProcess() throws Exception { GetMessageColumns getMessageColumns = new GetMessageColumns(); Set<String> sourceIds = getMessageColumns.getColumnNames(3); LOGGER.info(sourceIds.toString()); }