func exportProduct(export exportConfig, wg *sync.WaitGroup) {
	defer wg.Done()
	client := mixpanel.New(export.Product, export.Creds.Key, export.Creds.Secret)
	eventData := make(chan mixpanel.EventData)
	var chans []chan mixpanel.EventData
	makeChan := func() chan mixpanel.EventData {
		c := make(chan mixpanel.EventData, 100)
		chans = append(chans, c)
		return c
	}
	if cfg.JSON.State {
		c := makeChan()
		wg.Add(1)
		go func() {
			defer wg.Done()
			writer, cleanup := createExportFile(export, cfg.JSON, "", "json")
			defer cleanup()
			exports.JSONStreamer(writer, c)
		}()
	}
	if cfg.CSV.State {
		c := makeChan()
		wg.Add(1)
		go func() {
			defer wg.Done()
			writer, cleanup := createExportFile(export, cfg.CSV, "", "csv")
			defer cleanup()
			exports.CSVStreamer(writer, c)
		}()
	}
	if cfg.Columns.State {
		fp, err := os.Open(cfg.Columns.Columns)
		if err != nil {
			log.Fatalf("Couldn't open column defintions: %v", err)
		}
		columns := make(map[string]map[string][]string)
		if err := json.NewDecoder(fp).Decode(&columns); err != nil {
			log.Fatalf("Failed to read column definitions: %v", err)
		}
		fp.Close()
		if prodCols, ok := columns[export.Product]; ok {
			c := makeChan()
			wg.Add(1)
			go func() {
				defer wg.Done()
				defs := make(map[string]exports.EventColumnDef)
				for event, cols := range prodCols {
					writer, cleanup := createExportFile(
						export, cfg.Columns.fileExportConfig, event, "csv")
					defs[event] = exports.NewEventColumnDef(writer, cols)
					defer cleanup()
				}
				exports.CSVColumnStreamer(defs, c)
			}()
		}
	}
	go func() {
		defer close(eventData)
		total := 0
		end := export.End.AddDate(0, 0, 1)
		for date := export.Start; date.Before(end); date = date.AddDate(0, 0, 1) {
			num, err := client.ExportDate(date, eventData, nil)
			dateStr := date.Format("2006-01-02")
			if err != nil {
				log.Printf("%s: %s: export failed: %v", dateStr, export.Product, err)
				failedExports[export.Product] = true
				return
			} else if num == 0 {
				log.Printf("%s: %s: no records.", dateStr, export.Product)
			}
			total += num
		}
		log.Printf("%s: %d records.", export.Product, total)
	}()
	for data := range eventData {
		for _, ch := range chans {
			ch <- data
		}
	}
	for _, ch := range chans {
		close(ch)
	}
}
exportProduct is called once for each individual mixpanel product to be
exported. It starts each export function in its own goroutine and will block
until all events have been processed.
package main
import (
	"compress/gzip"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"path"
	"runtime"
	"runtime/pprof"
	"strings"
	"sync"
	"syscall"
	"time"
	"github.com/erik/mixport/exports"
	"github.com/erik/mixport/mixpanel"
	flag "github.com/ogier/pflag"
	"gopkg.in/gcfg.v1"
)
type mixpanelCredentials struct {
	Key    string
	Secret string
	Token  string
}
type fileExportConfig struct {
	State        bool
	Gzip         bool
	Fifo         bool
	RemoveFailed bool
	Directory    string
}
type columnExportConfig struct {
	fileExportConfig
	Columns string
}
type configFormat struct {
	Product map[string]*mixpanelCredentials
	JSON    fileExportConfig
	CSV     fileExportConfig
	Columns columnExportConfig
}
type exportConfig struct {
	Product    string
	Creds      mixpanelCredentials
	Start, End time.Time
}
var cfg = configFormat{}
var failedExports = make(map[string]bool)
func main() {
	flag.Usage = func() {
		fmt.Println(`Usage: mixport [OPTIONS]
Download and transform Mixpanel event data.
Options:
  -h, --help      Display this message.
  -c, --config    Path to configuration file, defaulting to "./mixport.conf"
  -d, --date      Date of data to pull in YYYY/MM/DD, defaulting to yesterday.
  -p, --procs     Maximum number of OS threads for the go runtime to use,
                  defaulting to number of CPUs available on the machine.
  --products      Comma separated list of products to export.
  --prof          Dump pprof info to the named file.
  -r, --range     Specify a date range to pull data for in YYYY/MM/DD-YYYY/MM/DD
                  format.`)
	}
	var (
		configFile  = flag.StringP("config", "c", "./mixport.conf", "")
		dateString  = flag.StringP("date", "d", "", "")
		rangeString = flag.StringP("range", "r", "", "")
		cpuProfile  = flag.String("prof", "", "")
		productList = flag.String("products", "", "")
		maxProcs    = flag.IntP("procs", "p", runtime.NumCPU(), "")
	)
	flag.Parse()
	runtime.GOMAXPROCS(*maxProcs)
	if *cpuProfile != "" {
		f, err := os.Create(*cpuProfile)
		if err != nil {
			log.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}
	if err := gcfg.ReadFileInto(&cfg, *configFile); err != nil {
		log.Fatalf("Failed to load %s: %s", *configFile, err)
	}
	var exportStart, exportEnd time.Time
	if *dateString == "" {
		year, month, day := time.Now().UTC().AddDate(0, 0, -1).Date()
		exportStart = time.Date(year, month, day, 0, 0, 0, 0, time.UTC)
		exportEnd = exportStart
	} else {
		if d, err := time.Parse("2006/01/02", *dateString); err != nil {
			log.Fatalf("Invalid date: %s, should be in YYYY/MM/DD format",
				*dateString)
		} else {
			exportStart = d
			exportEnd = exportStart
		}
	}
	if *rangeString != "" {
		formatError := func() {
			log.Fatalf("Invalid range: %s, should be YYYY/MM/DD-YYYY/MM/DD.", *rangeString)
		}
		parts := strings.Split(*rangeString, "-")
		if len(parts) != 2 {
			formatError()
		}
		var err error
		if exportStart, err = time.Parse("2006/01/02", parts[0]); err != nil {
			formatError()
		}
		if exportEnd, err = time.Parse("2006/01/02", parts[1]); err != nil {
			formatError()
		}
	}
	fileConfigs := []fileExportConfig{cfg.CSV, cfg.JSON, cfg.Columns.fileExportConfig}
	for _, config := range fileConfigs {
		if config.State && config.Fifo && config.RemoveFailed {
			log.Fatalf("Can't have both `fifo=true` and `removefailed=true`")
		}
	}
	products := make(map[string]*mixpanelCredentials)
	if *productList == "" {
		products = cfg.Product
	} else {
		for _, name := range strings.Split(*productList, ",") {
			if creds, ok := cfg.Product[name]; ok {
				products[name] = creds
			} else {
				log.Fatalf("Don't have credentials specified for %s", name)
			}
		}
	}
	var wg sync.WaitGroup
	for product, creds := range products {
		wg.Add(1)
		go exportProduct(exportConfig{product, *creds, exportStart, exportEnd}, &wg)
	}
	wg.Wait()
	if len(failedExports) > 0 {
		log.Printf("Finished with errors:")
		for product := range failedExports {
			log.Printf("\t%s", product)
		}
		os.Exit(1)
	}
}
func createExportFile(export exportConfig, conf fileExportConfig, event, ext string) (io.Writer, func()) {
	if conf.Gzip {
		ext += ".gz"
	}
	start, end := export.Start, export.End
	timeFmt := "20060102"
	stamp := start.Format(timeFmt)
	if start != end {
		stamp += fmt.Sprintf("-%s", end.Format(timeFmt))
	}
	name := path.Join(conf.Directory, export.Product)
	if event != "" {
		name += fmt.Sprintf("-%s", event)
	}
	name += fmt.Sprintf("-%s.%s", stamp, ext)
	if conf.Fifo {
		if err := syscall.Mkfifo(name, syscall.S_IRWXU); err != nil {
			log.Fatalf("Couldn't create named pipe: %s", err)
		}
	}
	fp, err := os.Create(name)
	if err != nil {
		log.Fatalf("Couldn't create file: %s", err)
	}
	writer := (io.Writer)(fp)
	if conf.Gzip {
		writer = gzip.NewWriter(fp)
	}
	cleanup := func() {
		if conf.Gzip {
			(writer).(*gzip.Writer).Close()
		}
		fp.Close()
		if conf.Fifo {
			os.Remove(name)
		}
		if conf.RemoveFailed {
			if _, err := failedExports[export.Product]; err {
				os.Remove(name)
			}
		}
	}
	return writer, cleanup
}