public static TStream<JsonObject> waterDetector(Topology topology,int wellId){
  Random rNum=new Random();
  TStream<Integer> temp=topology.poll(null,1,TimeUnit.SECONDS);
  TStream<Integer> acidity=topology.poll(null,1,TimeUnit.SECONDS);
  TStream<Integer> ecoli=topology.poll(null,1,TimeUnit.SECONDS);
  TStream<Integer> lead=topology.poll(null,1,TimeUnit.SECONDS);
  TStream<Integer> id=topology.poll(null,1,TimeUnit.SECONDS);
  temp.tag("temperature","well" + wellId);
  acidity.tag("acidity","well" + wellId);
  ecoli.tag("ecoli","well" + wellId);
  lead.tag("lead","well" + wellId);
  id.tag("well" + wellId);
  TStream<JsonObject> tempObj=temp.map(null);
  TStream<JsonObject> acidityObj=acidity.map(null);
  TStream<JsonObject> ecoliObj=ecoli.map(null);
  TStream<JsonObject> leadObj=lead.map(null);
  TStream<JsonObject> idObj=id.map(null);
  HashSet<TStream<JsonObject>> set=new HashSet<TStream<JsonObject>>();
  set.add(acidityObj);
  set.add(acidityObj);
  set.add(ecoliObj);
  set.add(leadObj);
  set.add(idObj);
  TStream<JsonObject> allReadings=tempObj.union(set);
  return allReadings;
}

public static TStream < JsonObject > waterDetector ( Topology topology , int wellId ) { Random rNum = new Random ( ) ; TStream < Integer > temp = topology . poll ( null , 1 , TimeUnit . SECONDS ) ; TStream < Integer > acidity = topology . poll ( null , 1 , TimeUnit . SECONDS ) ; TStream < Integer > ecoli = topology . poll ( null , 1 , TimeUnit . SECONDS ) ; TStream < Integer > lead = topology . poll ( null , 1 , TimeUnit . SECONDS ) ; TStream < Integer > id = topology . poll ( null , 1 , TimeUnit . SECONDS ) ; temp . tag ( "temperature" , "well" + wellId ) ; acidity . tag ( "acidity" , "well" + wellId ) ; ecoli . tag ( "ecoli" , "well" + wellId ) ; lead . tag ( "lead" , "well" + wellId ) ; id . tag ( "well" + wellId ) ; TStream < JsonObject > tempObj = temp . map ( null ) ; TStream < JsonObject > acidityObj = acidity . map ( null ) ; TStream < JsonObject > ecoliObj = ecoli . map ( null ) ; TStream < JsonObject > leadObj = lead . map ( null ) ; TStream < JsonObject > idObj = id . map ( null ) ; HashSet < TStream < JsonObject > > set = new HashSet < TStream < JsonObject > > ( ) ; set . add ( acidityObj ) ; set . add ( acidityObj ) ; set . add ( ecoliObj ) ; set . add ( leadObj ) ; set . add ( idObj ) ; TStream < JsonObject > allReadings = tempObj . union ( set ) ; return allReadings ; }
creates a tstream & lt ; jsonobject & gt ; for each sensor reading for each well . unions all the tstream & lt ; jsonobject & gt ; into a single one representing all readings on the well .
