// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFNodeInputFormat.java
// public class OSMPBFNodeInputFormat extends OSMPBFInputFormat<NodeEntity> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new NodeParser(p);
//     }
// 
// }
// 
// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFRelationInputFormat.java
// public class OSMPBFRelationInputFormat extends OSMPBFInputFormat<Relation> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new RelationParser(p);
//     }
// 
// }
// 
// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFWayInputFormat.java
// public class OSMPBFWayInputFormat extends OSMPBFInputFormat<WayEntity> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new WayParser(p);
//     }
// 
// }


import com.sun.org.apache.xpath.internal.operations.Bool;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFNodeInputFormat;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFRelationInputFormat;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFWayInputFormat;
import io.sharedstreets.tools.builder.osm.model.*;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Meter;
import org.apache.flink.util.Collector;
import java.util.ArrayList;
                    return false;
            }
        });

        // link way id to node ids with ordering
        // way_id, node_id, order, terminating
        unfilteredOrderedWayNodeLink = filteredWays
                .flatMap(new FlatMapFunction<WayEntity, Tuple4<Long, Long, Integer, Boolean>>() {
                    @Override
                    public void flatMap(WayEntity value, Collector<Tuple4<Long, Long, Integer, Boolean>> out) throws Exception {

                        if (value.relatedObjects != null) {
                            int c = 0;
                            int max = value.relatedObjects.length - 1;
                            for (RelatedObject r : value.relatedObjects) {
                                boolean terminal_point = false;
                                if (c == 0 || c == max)
                                    terminal_point = true;

                                out.collect(new Tuple4<Long, Long, Integer, Boolean>(value.id, r.relatedId, c++, terminal_point));
                            }
                        }
                    }
                }).partitionByHash(0);
        
        // join ways with node positions
        // way_id, order, node_id, lat, lon
        DataSet<Tuple5<Long, Integer, Long, Double, Double>> joinedWaysWithPoints = unfilteredOrderedWayNodeLink
                .joinWithHuge(nodePositions)
                .where(1)
// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFNodeInputFormat.java
// public class OSMPBFNodeInputFormat extends OSMPBFInputFormat<NodeEntity> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new NodeParser(p);
//     }
// 
// }
// 
// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFRelationInputFormat.java
// public class OSMPBFRelationInputFormat extends OSMPBFInputFormat<Relation> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new RelationParser(p);
//     }
// 
// }
// 
// Path: src/main/java/io/sharedstreets/tools/builder/osm/inputs/OSMPBFWayInputFormat.java
// public class OSMPBFWayInputFormat extends OSMPBFInputFormat<WayEntity> {
// 
//     @Override
//     protected Parser createParser(Osmformat.PrimitiveBlock p) {
//         return new WayParser(p);
//     }
// 
// }


// Path: src/main/java/io/sharedstreets/tools/builder/osm/OSMDataStream.java
import com.sun.org.apache.xpath.internal.operations.Bool;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFNodeInputFormat;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFRelationInputFormat;
import io.sharedstreets.tools.builder.osm.inputs.OSMPBFWayInputFormat;
import io.sharedstreets.tools.builder.osm.model.*;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Meter;
import org.apache.flink.util.Collector;
import java.util.ArrayList;

                    return false;
            }
        });

        // link way id to node ids with ordering
        // way_id, node_id, order, terminating
        unfilteredOrderedWayNodeLink = filteredWays
                .flatMap(new FlatMapFunction<WayEntity, Tuple4<Long, Long, Integer, Boolean>>() {
                    @Override
                    public void flatMap(WayEntity value, Collector<Tuple4<Long, Long, Integer, Boolean>> out) throws Exception {

                        if (value.relatedObjects != null) {
                            int c = 0;
                            int max = value.relatedObjects.length - 1;
                            for (RelatedObject r : value.relatedObjects) {
                                boolean terminal_point = false;
                                if (c == 0 || c == max)
                                    terminal_point = true;

                                out.collect(new Tuple4<Long, Long, Integer, Boolean>(value.id, r.relatedId, c++, terminal_point));
                            }
                        }
                    }
                }).partitionByHash(0);
        
        // join ways with node positions
        // way_id, order, node_id, lat, lon
        DataSet<Tuple5<Long, Integer, Long, Double, Double>> joinedWaysWithPoints = unfilteredOrderedWayNodeLink
                .joinWithHuge(nodePositions)
                .where(1)
                .equalTo(0).map(new MapFunction<Tuple2<Tuple4<Long, Long, Integer, Boolean>, Tuple3<Long, Double, Double>>, Tuple5<Long, Integer, Long, Double, Double>>() {