func (g *joinGroup) Collect(src int, p timeMessage) error {
	t := p.Time().Round(g.n.j.Tolerance)
	if t.Before(g.oldestTime) || g.oldestTime.IsZero() {
		g.oldestTime = t
	}
	var set *joinset
	sets := g.sets[t]
	if len(sets) == 0 {
		set = g.newJoinset(t)
		sets = append(sets, set)
		g.sets[t] = sets
	}
	for i := 0; i < len(sets); i++ {
		if !sets[i].Has(src) {
			set = sets[i]
			break
		}
	}
	if set == nil {
		set = g.newJoinset(t)
		sets = append(sets, set)
		g.sets[t] = sets
	}
	set.Set(src, p)
	g.head[src] = t
	onlyReadySets := false
	for _, t := range g.head {
		if !t.After(g.oldestTime) {
			onlyReadySets = true
			break
		}
	}
	err := g.emit(onlyReadySets)
	if err != nil {
		return err
	}
	return nil
}
Collect a point from a given parent.
emit the oldest set if we have collected enough data.
package kapacitor
import (
	"fmt"
	"log"
	"sync"
	"time"
	"github.com/influxdata/influxdb/influxql"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/expvar"
	"github.com/influxdata/kapacitor/models"
	"github.com/influxdata/kapacitor/pipeline"
	"github.com/pkg/errors"
)
type JoinNode struct {
	node
	j         *pipeline.JoinNode
	fill      influxql.FillOption
	fillValue interface{}
	groupsMu sync.RWMutex
	groups   map[models.GroupID]*joinGroup
	lowMarks map[srcGroup]time.Time
	matchGroupsBuffer map[models.GroupID][]srcPoint
	specificGroupsBuffer map[models.GroupID][]srcPoint
	reported    map[int]bool
	allReported bool
}
func newJoinNode(et *ExecutingTask, n *pipeline.JoinNode, l *log.Logger) (*JoinNode, error) {
	jn := &JoinNode{
		j:                    n,
		node:                 node{Node: n, et: et, logger: l},
		groups:               make(map[models.GroupID]*joinGroup),
		matchGroupsBuffer:    make(map[models.GroupID][]srcPoint),
		specificGroupsBuffer: make(map[models.GroupID][]srcPoint),
		lowMarks:             make(map[srcGroup]time.Time),
		reported:             make(map[int]bool),
	}
	switch fill := n.Fill.(type) {
	case string:
		switch fill {
		case "null":
			jn.fill = influxql.NullFill
		case "none":
			jn.fill = influxql.NoFill
		default:
			return nil, fmt.Errorf("unexpected fill option %s", fill)
		}
	case int64, float64:
		jn.fill = influxql.NumberFill
		jn.fillValue = fill
	default:
		jn.fill = influxql.NoFill
	}
	jn.node.runF = jn.runJoin
	return jn, nil
}
func (n *JoinNode) runJoin([]byte) error {
	consumer := edge.NewMultiConsumerWithStats(n.ins, n)
	valueF := func() int64 {
		n.groupsMu.RLock()
		l := len(n.groups)
		n.groupsMu.RUnlock()
		return int64(l)
	}
	n.statMap.Set(statCardinalityGauge, expvar.NewIntFuncGauge(valueF))
	return consumer.Consume()
}
func (n *JoinNode) BufferedBatch(src int, batch edge.BufferedBatchMessage) error {
	return n.doMessage(src, batch)
}
func (n *JoinNode) Point(src int, p edge.PointMessage) error {
	return n.doMessage(src, p)
}
func (n *JoinNode) Barrier(src int, b edge.BarrierMessage) error {
	return edge.Forward(n.outs, b)
}
func (n *JoinNode) Finish() error {
	for _, group := range n.groups {
		if err := group.Finish(); err != nil {
			return err
		}
	}
	return nil
}
type messageMeta interface {
	edge.Message
	edge.PointMeta
}
type srcPoint struct {
	Src int
	Msg messageMeta
}
func (n *JoinNode) doMessage(src int, m messageMeta) error {
	n.timer.Start()
	defer n.timer.Stop()
	if len(n.j.Dimensions) > 0 {
		n.matchPoints(srcPoint{Src: src, Msg: m})
	} else {
		group := n.getOrCreateGroup(m.GroupID())
		group.Collect(src, m)
	}
	return nil
}
func (n *JoinNode) matchPoints(p srcPoint) {
	if !n.allReported {
		n.reported[p.Src] = true
		n.allReported = len(n.reported) == len(n.ins)
	}
	t := p.Msg.Time().Round(n.j.Tolerance)
	groupId := models.ToGroupID(
		p.Msg.Name(),
		p.Msg.GroupInfo().Tags,
		models.Dimensions{
			ByName:   p.Msg.Dimensions().ByName,
			TagNames: n.j.Dimensions,
		},
	)
	srcG := srcGroup{src: p.Src, groupId: groupId}
	n.lowMarks[srcG] = t
	var lowMark time.Time
	if n.allReported {
		for s := 0; s < len(n.ins); s++ {
			sg := srcGroup{src: s, groupId: groupId}
			if lm := n.lowMarks[sg]; lowMark.IsZero() || lm.Before(lowMark) {
				lowMark = lm
			}
		}
	}
	if n.allReported {
		var i int
		buf := n.specificGroupsBuffer[groupId]
		l := len(buf)
		for i = 0; i < l; i++ {
			st := buf[i].Msg.Time().Round(n.j.Tolerance)
			if st.Before(lowMark) {
				n.sendSpecificPoint(buf[i])
			} else {
				break
			}
		}
		n.specificGroupsBuffer[groupId] = buf[i:]
	}
	if len(p.Msg.Dimensions().TagNames) > len(n.j.Dimensions) {
		matches := n.matchGroupsBuffer[groupId]
		matched := false
		var i int
		l := len(matches)
		for i = 0; i < l; i++ {
			match := matches[i]
			pt := match.Msg.Time().Round(n.j.Tolerance)
			if pt.Equal(t) {
				n.sendMatchPoint(p, match)
				matched = true
			}
			if !pt.Before(lowMark) {
				break
			}
		}
		if n.allReported {
			n.matchGroupsBuffer[groupId] = matches[i:]
		}
		if !matched {
			if n.allReported && t.Before(lowMark) {
				n.sendSpecificPoint(p)
			} else {
				n.specificGroupsBuffer[groupId] = append(n.specificGroupsBuffer[groupId], p)
			}
		}
	} else {
		n.matchGroupsBuffer[groupId] = append(n.matchGroupsBuffer[groupId], p)
		var i int
		buf := n.specificGroupsBuffer[groupId]
		l := len(buf)
		for i = 0; i < l; i++ {
			st := buf[i].Msg.Time().Round(n.j.Tolerance)
			if st.Equal(t) {
				n.sendMatchPoint(buf[i], p)
			} else {
				break
			}
		}
		n.specificGroupsBuffer[groupId] = buf[i:]
	}
}
func (n *JoinNode) sendMatchPoint(specific, matched srcPoint) {
	var newMatched messageMeta
	switch msg := matched.Msg.(type) {
	case edge.BufferedBatchMessage:
		b := msg.ShallowCopy()
		b.SetBegin(b.Begin().ShallowCopy())
		b.Begin().SetTags(specific.Msg.GroupInfo().Tags)
		newMatched = b
	case edge.PointMessage:
		p := msg.ShallowCopy()
		info := specific.Msg.GroupInfo()
		p.SetTagsAndDimensions(info.Tags, info.Dimensions)
		newMatched = p
	}
	group := n.getOrCreateGroup(specific.Msg.GroupID())
	group.Collect(specific.Src, specific.Msg)
	group.Collect(matched.Src, newMatched)
}
func (n *JoinNode) sendSpecificPoint(specific srcPoint) {
	group := n.getOrCreateGroup(specific.Msg.GroupID())
	group.Collect(specific.Src, specific.Msg)
}
func (n *JoinNode) getOrCreateGroup(groupID models.GroupID) *joinGroup {
	group := n.groups[groupID]
	if group == nil {
		group = n.newGroup(len(n.ins))
		n.groupsMu.Lock()
		n.groups[groupID] = group
		n.groupsMu.Unlock()
	}
	return group
}
func (n *JoinNode) newGroup(count int) *joinGroup {
	return &joinGroup{
		n:    n,
		sets: make(map[time.Time][]*joinset),
		head: make([]time.Time, count),
	}
}
type joinGroup struct {
	n *JoinNode
	sets       map[time.Time][]*joinset
	head       []time.Time
	oldestTime time.Time
}
func (g *joinGroup) Finish() error {
	return g.emitAll()
}