From 94bebbd276d5ef2f00a5ad9fbde57472e9c4b2a6 Mon Sep 17 00:00:00 2001
From: Eric Fischer <enf@pobox.com>
Date: Thu, 1 Dec 2016 12:04:49 -0800
Subject: [PATCH] Write GeoJSON to the filter and read (but don't parse) what
 comes back

---
 Makefile   |   2 +-
 plugin.cpp | 131 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 plugin.hpp |   1 +
 tile.cpp   |   1 +
 4 files changed, 134 insertions(+), 1 deletion(-)

diff --git a/Makefile b/Makefile
index 0dc855b..b72f179 100644
--- a/Makefile
+++ b/Makefile
@@ -50,7 +50,7 @@ tippecanoe-enumerate: enumerate.o
 	$(CXX) $(PG) $(LIBS) $(FINAL_FLAGS) $(CFLAGS) -o $@ $^ $(LDFLAGS) -lsqlite3
 
 tippecanoe-decode: decode.o projection.o mvt.o plugin.o
-	$(CXX) $(PG) $(LIBS) $(FINAL_FLAGS) $(CXXFLAGS) -o $@ $^ $(LDFLAGS) -lm -lz -lsqlite3
+	$(CXX) $(PG) $(LIBS) $(FINAL_FLAGS) $(CXXFLAGS) -o $@ $^ $(LDFLAGS) -lm -lz -lsqlite3 -lpthread
 
 tile-join: tile-join.o projection.o pool.o mbtiles.o mvt.o memfile.o
 	$(CXX) $(PG) $(LIBS) $(FINAL_FLAGS) $(CXXFLAGS) -o $@ $^ $(LDFLAGS) -lm -lz -lsqlite3 -lpthread
diff --git a/plugin.cpp b/plugin.cpp
index afde545..aa0fc11 100644
--- a/plugin.cpp
+++ b/plugin.cpp
@@ -3,11 +3,142 @@
 #include <vector>
 #include <string>
 #include <map>
+#include <pthread.h>
+#include <unistd.h>
 #include "mvt.hpp"
 #include "plugin.hpp"
 #include "projection.hpp"
 #include "geometry.hpp"
 
+struct writer_arg {
+	int *pipe_orig;
+	mvt_layer *layer;
+	unsigned z;
+	unsigned x;
+	unsigned y;
+};
+
+void *run_writer(void *a) {
+	writer_arg *wa = (writer_arg *) a;
+
+	// XXX worry about SIGPIPE?
+
+	FILE *fp = fdopen(wa->pipe_orig[1], "w");
+	if (fp == NULL) {
+		perror("fdopen (pipe writer)");
+		exit(EXIT_FAILURE);
+	}
+
+	layer_to_geojson(fp, *(wa->layer), wa->z, wa->x, wa->y);
+
+	if (fclose(fp) != 0) {
+		perror("fclose output to filter");
+		exit(EXIT_FAILURE);
+	}
+
+	return NULL;
+}
+
+mvt_layer filter_layer(const char *filter, mvt_layer &layer, unsigned z, unsigned x, unsigned y) {
+	// This will create two pipes, a new thread, and a new process.
+	//
+	// The new process will read from one pipe and write to the other, and execute the filter.
+	// The new thread will write the GeoJSON to the pipe that leads to the filter.
+	// The original thread will read the GeoJSON from the filter and convert it back into vector tiles.
+
+	int pipe_orig[2], pipe_filtered[2];
+	if (pipe(pipe_orig) < 0) {
+		perror("pipe (original features)");
+		exit(EXIT_FAILURE);
+	}
+	if (pipe(pipe_filtered) < 0) {
+		perror("pipe (filtered features)");
+		exit(EXIT_FAILURE);
+	}
+
+	pid_t pid = fork();
+	if (pid < 0) {
+		perror("fork");
+		exit(EXIT_FAILURE);
+	} else if (pid == 0) {
+		// child
+
+		if (dup2(pipe_orig[0], 0) < 0) {
+			perror("dup child stdin");
+			exit(EXIT_FAILURE);
+		}
+		if (dup2(pipe_filtered[1], 1) < 0) {
+			perror("dup child stdout");
+			exit(EXIT_FAILURE);
+		}
+		if (close(pipe_orig[1]) != 0) {
+			perror("close output to filter");
+			exit(EXIT_FAILURE);
+		}
+		if (close(pipe_filtered[0]) != 0) {
+			perror("close input from filter");
+			exit(EXIT_FAILURE);
+		}
+
+		// XXX close other fds?
+
+		// XXX add zyx args
+		if (execlp("sh", "sh", "-c", filter, NULL) != 0) {
+			perror("exec");
+			exit(EXIT_FAILURE);
+		}
+	} else {
+		// parent
+
+		if (close(pipe_orig[0]) != 0) {
+			perror("close filter-side reader");
+			exit(EXIT_FAILURE);
+		}
+		if (close(pipe_filtered[1]) != 0) {
+			perror("close filter-side writer");
+			exit(EXIT_FAILURE);
+		}
+
+		writer_arg wa;
+		wa.pipe_orig = pipe_orig;
+		wa.layer = &layer;
+		wa.z = z;
+		wa.x = x;
+		wa.y = y;
+
+		pthread_t writer;
+		if (pthread_create(&writer, NULL, run_writer, &wa) != 0) {
+			perror("pthread_create (filter writer)");
+			exit(EXIT_FAILURE);
+		}
+
+		char buf[200];
+		size_t count;
+		while ((count = read(pipe_filtered[0], buf, 200)) != 0) {
+			write(1, buf, count);
+		}
+
+		int stat_loc;
+		if (waitpid(pid, &stat_loc, 0) < 0) {
+			perror("waitpid for filter\n");
+			exit(EXIT_FAILURE);
+		}
+
+		if (close(pipe_filtered[0]) != 0) {
+			perror("close output from filter");
+			exit(EXIT_FAILURE);
+		}
+
+		void *ret;
+		if (pthread_join(writer, &ret) != 0) {
+			perror("pthread_join filter writer");
+			exit(EXIT_FAILURE);
+		}
+	}
+
+	return layer;
+}
+
 struct lonlat {
 	int op;
 	double lon;
diff --git a/plugin.hpp b/plugin.hpp
index 1308006..0981ce0 100644
--- a/plugin.hpp
+++ b/plugin.hpp
@@ -1,2 +1,3 @@
 void layer_to_geojson(FILE *fp, mvt_layer &layer, unsigned z, unsigned x, unsigned y);
+mvt_layer filter_layer(const char *filter, mvt_layer &layer, unsigned z, unsigned x, unsigned y);
 void fprintq(FILE *f, const char *s);
diff --git a/tile.cpp b/tile.cpp
index ae34f16..0ab9f5b 100644
--- a/tile.cpp
+++ b/tile.cpp
@@ -29,6 +29,7 @@
 #include "serial.hpp"
 #include "options.hpp"
 #include "main.hpp"
+#include "plugin.hpp"
 
 #define CMD_BITS 3