func createFile(ctx context.Context, path string) (io.Writer, func()) {
	out, err := file.Create(ctx, path)
	if err != nil {
		log.Panicf("create %s: %v", path, err)
	}
	log.Printf("Creating %v", path)
	buf := bufio.NewWriterSize(out.Writer(ctx), 1<<20)
	return buf, func() {
		if err := buf.Flush(); err != nil {
			log.Panicf("close %s: %v", path, err)
		}
		if err := out.Close(ctx); err != nil {
			log.Panicf("close %s: %v", path, err)
		}
	}
}
Open a buffered file writer. The 2nd arg (func) must be called to flush &
close the file.
package cmd
import (
	"bufio"
	"bytes"
	"context"
	"encoding/gob"
	"io"
	"log"
	"github.com/grailbio/base/file"
	"github.com/grailbio/base/recordio"
	"github.com/grailbio/base/recordio/recordiozstd"
	"github.com/grailbio/bio/fusion"
)
type fusionWriter struct {
	out    file.File
	w      recordio.Writer
	geneDB *fusion.GeneDB
	opts   fusion.Opts
}
const (
	fileVersionHeader = "af4version"
	fileVersion       = "AF4_V1"
)
type fusionFileHeader struct {
	Opts fusion.Opts
	Gene []fusion.GeneInfo
}
func encodeGOB(gw *gob.Encoder, v interface{}) {
	if err := gw.Encode(v); err != nil {
		panic(err)
	}
}
func decodeGOB(gr *gob.Decoder, v interface{}) {
	if err := gr.Decode(v); err != nil {
		panic(err)
	}
}
func newFusionWriter(ctx context.Context, outPath string, geneDB *fusion.GeneDB, opts fusion.Opts) *fusionWriter {
	recordiozstd.Init()
	out, err := file.Create(ctx, outPath)
	if err != nil {
		log.Panicf("rio open %v: %v", outPath, err)
	}
	w := recordio.NewWriter(out.Writer(ctx), recordio.WriterOpts{
		Transformers: []string{recordiozstd.Name},
	})
	w.AddHeader(fileVersionHeader, fileVersion)
	w.AddHeader(recordio.KeyTrailer, true)
	return &fusionWriter{out: out, w: w, geneDB: geneDB, opts: opts}
}
func (w *fusionWriter) Write(c fusion.Candidate) {
	b := bytes.NewBuffer(nil)
	gw := gob.NewEncoder(b)
	encodeGOB(gw, c)
	w.w.Append(b.Bytes())
}
func (w *fusionWriter) Close(ctx context.Context) {
	b := bytes.NewBuffer(nil)
	gw := gob.NewEncoder(b)
	h := fusionFileHeader{
		Opts: w.opts,
	}
	minGeneID, limitGeneID := w.geneDB.GeneIDRange()
	for gid := minGeneID; gid < limitGeneID; gid++ {
		h.Gene = append(h.Gene, *w.geneDB.GeneInfo(gid))
	}
	encodeGOB(gw, h)
	w.w.SetTrailer(b.Bytes())
	err := w.w.Finish()
	if err != nil {
		log.Panic("close", err)
	}
	if err := w.out.Close(ctx); err != nil {
		log.Panic("close", err)
	}
}
type fusionReader struct {
	in     file.File
	r      recordio.Scanner
	geneDB *fusion.GeneDB
	opts   fusion.Opts
	c fusion.Candidate  
}
func newFusionReader(ctx context.Context, inPath string) *fusionReader {
	in, err := file.Open(ctx, inPath)
	if err != nil {
		log.Panicf("open %s: %v", inPath, err)
	}
	recordiozstd.Init()
	r := recordio.NewScanner(in.Reader(ctx), recordio.ScannerOpts{})
	versionFound := false
	for _, kv := range r.Header() {
		if kv.Key == fileVersionHeader {
			if kv.Value.(string) != fileVersion {
				log.Panicf("TODO: fusion file version mismatch, got %v, expect %v",
					kv.Value.(string), fileVersion)
			}
			versionFound = true
			break
		}
	}
	if !versionFound {
		log.Panic(fileVersionHeader + " not found")
	}
	gr := gob.NewDecoder(bytes.NewReader(r.Trailer()))
	h := fusionFileHeader{}
	decodeGOB(gr, &h)
	geneDB := fusion.NewGeneDB(h.Opts)
	geneDB.PrepopulateGeneInfo(h.Gene)
	return &fusionReader{in: in, r: r, geneDB: geneDB, opts: h.Opts}
}
func (r *fusionReader) Opts() fusion.Opts { return r.opts }
func (r *fusionReader) GeneDB() *fusion.GeneDB { return r.geneDB }
func (r *fusionReader) Scan() bool {
	if !r.r.Scan() {
		return false
	}
	b := bytes.NewReader(r.r.Get().([]byte))
	gr := gob.NewDecoder(b)
	r.c = fusion.Candidate{}
	decodeGOB(gr, &r.c)
	return true
}
func (r *fusionReader) Get() fusion.Candidate { return r.c }
func (r *fusionReader) Close(ctx context.Context) {
	if err := r.r.Err(); err != nil {
		log.Panic(err)
	}
	if err := r.in.Close(ctx); err != nil {
		log.Panic(err)
	}
}