// open an archive for writing batch recordings
func (r *Service) newBatchArchive(rid uuid.UUID) (*batchArchive, error) {
	rpath := path.Join(r.saveDir, rid.String()+batchEXT)
	f, err := os.Create(rpath)
	if err != nil {
		return nil, err
	}
	archive := zip.NewWriter(f)
	return &batchArchive{f: f, archive: archive}, nil
}

// create new stream writer
func (r *Service) newStreamWriter(rid uuid.UUID) (io.WriteCloser, error) {
	rpath := path.Join(r.saveDir, rid.String()+streamEXT)
	f, err := os.Create(rpath)
	if err != nil {
		return nil, fmt.Errorf("failed to save recording: %s", err)
	}
	gz := gzip.NewWriter(f)
	sw := streamWriter{f: f, gz: gz}
	return sw, nil
}

// Record the stream for a duration
func (r *Service) doRecordStream(rid uuid.UUID, dur time.Duration, dbrps []kapacitor.DBRP, started chan struct{}) error {
	e, err := r.TaskMaster.NewFork(rid.String(), dbrps)
	if err != nil {
		return err
	}
	sw, err := r.newStreamWriter(rid)
	if err != nil {
		return err
	}
	defer sw.Close()

	done := make(chan struct{})
	go func() {
		close(started)
		start := time.Time{}
		closed := false
		for p, ok := e.NextPoint(); ok; p, ok = e.NextPoint() {
			if closed {
				continue
			}
			if start.IsZero() {
				start = p.Time
			}
			if p.Time.Sub(start) > dur {
				closed = true
				close(done)
				//continue to read any data already on the edge, but just drop it.
				continue
			}
			kapacitor.WritePointForRecording(sw, p, precision)
		}
	}()
	<-done
	e.Abort()
	r.TaskMaster.DelFork(rid.String())
	return nil
}