func (w *windowByTime) batch(tmax time.Time) edge.BufferedBatchMessage {
	points := w.buf.points()
	return edge.NewBufferedBatchMessage(
		edge.NewBeginBatchMessage(
			w.name,
			w.group.Tags,
			w.group.Dimensions.ByName,
			tmax,
			len(points),
		),
		points,
		edge.NewEndBatchMessage(),
	)
}
batch returns the current window buffer as a batch message.
TODO(nathanielc): A possible optimization could be to not buffer the data at all if we know that we do not have overlapping windows.
package kapacitor
import (
	"errors"
	"fmt"
	"log"
	"time"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/models"
	"github.com/influxdata/kapacitor/pipeline"
)
type WindowNode struct {
	node
	w *pipeline.WindowNode
}
func newWindowNode(et *ExecutingTask, n *pipeline.WindowNode, l *log.Logger) (*WindowNode, error) {
	if n.Period == 0 && n.PeriodCount == 0 {
		return nil, errors.New("window node must have either a non zero period or non zero period count")
	}
	wn := &WindowNode{
		w:    n,
		node: node{Node: n, et: et, logger: l},
	}
	wn.node.runF = wn.runWindow
	return wn, nil
}
func (n *WindowNode) runWindow([]byte) error {
	consumer := edge.NewGroupedConsumer(n.ins[0], n)
	n.statMap.Set(statCardinalityGauge, consumer.CardinalityVar())
	return consumer.Consume()
}
func (n *WindowNode) NewGroup(group edge.GroupInfo, first edge.PointMeta) (edge.Receiver, error) {
	r, err := n.newWindow(group, first)
	if err != nil {
		return nil, err
	}
	return edge.NewReceiverFromForwardReceiverWithStats(
		n.outs,
		edge.NewTimedForwardReceiver(n.timer, r),
	), nil
}
func (n *WindowNode) DeleteGroup(group models.GroupID) {
}
func (n *WindowNode) newWindow(group edge.GroupInfo, first edge.PointMeta) (edge.ForwardReceiver, error) {
	switch {
	case n.w.Period != 0:
		return newWindowByTime(
			first.Name(),
			first.Time(),
			group,
			n.w.Period,
			n.w.Every,
			n.w.AlignFlag,
			n.w.FillPeriodFlag,
			n.logger,
		), nil
	case n.w.PeriodCount != 0:
		return newWindowByCount(
			first.Name(),
			group,
			int(n.w.PeriodCount),
			int(n.w.EveryCount),
			n.w.FillPeriodFlag,
			n.logger,
		), nil
	default:
		return nil, errors.New("unreachable code, window node should have a non-zero period or period count")
	}
}
type windowByTime struct {
	name  string
	group edge.GroupInfo
	nextEmit time.Time
	buf *windowTimeBuffer
	align,
	fillPeriod bool
	period time.Duration
	every  time.Duration
	logger *log.Logger
}
func newWindowByTime(
	name string,
	t time.Time,
	group edge.GroupInfo,
	period,
	every time.Duration,
	align,
	fillPeriod bool,
	logger *log.Logger,
) *windowByTime {
	var nextEmit time.Time
	if fillPeriod {
		nextEmit = t.Add(period)
		if align {
			firstPeriod := nextEmit
			nextEmit = nextEmit.Truncate(every)
			if !nextEmit.After(firstPeriod) {
				nextEmit = nextEmit.Add(every)
			}
		}
	} else {
		nextEmit = t.Add(every)
		if align {
			nextEmit = nextEmit.Truncate(every)
		}
	}
	return &windowByTime{
		name:       name,
		group:      group,
		nextEmit:   nextEmit,
		buf:        &windowTimeBuffer{logger: logger},
		align:      align,
		fillPeriod: fillPeriod,
		period:     period,
		every:      every,
		logger:     logger,
	}
}
func (w *windowByTime) BeginBatch(edge.BeginBatchMessage) (edge.Message, error) {
	return nil, errors.New("window does not support batch data")
}
func (w *windowByTime) BatchPoint(edge.BatchPointMessage) (edge.Message, error) {
	return nil, errors.New("window does not support batch data")
}
func (w *windowByTime) EndBatch(edge.EndBatchMessage) (edge.Message, error) {
	return nil, errors.New("window does not support batch data")
}
func (w *windowByTime) Barrier(b edge.BarrierMessage) (edge.Message, error) {
	return b, nil
}
func (w *windowByTime) DeleteGroup(d edge.DeleteGroupMessage) (edge.Message, error) {
	return d, nil
}
func (w *windowByTime) Point(p edge.PointMessage) (msg edge.Message, err error) {
	if w.every == 0 {
		w.buf.insert(p)
		if !p.Time().Before(w.nextEmit) {
			oldest := p.Time().Add(-1 * w.period)
			w.buf.purge(oldest, false)
			msg = w.batch(p.Time())
			w.nextEmit = p.Time()
		}
	} else {
		if !p.Time().Before(w.nextEmit) {
			oldest := w.nextEmit.Add(-1 * w.period)
			w.buf.purge(oldest, true)
			msg = w.batch(w.nextEmit)
			w.nextEmit = p.Time().Add(w.every)
			if w.align {
				w.nextEmit = w.nextEmit.Truncate(w.every)
			}
		}
		w.buf.insert(p)
	}
	return
}