config_snapshot_simple_emitter.sk.go

// Code generated by solo-kit. DO NOT EDIT.

package v1

import (
	"context"
	"fmt"
	"time"

	gloo_solo_io "github.com/solo-io/gloo/projects/gloo/pkg/api/v1"
	github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes "github.com/solo-io/solo-kit/pkg/api/v1/resources/common/kubernetes"

	"go.opencensus.io/stats"

	"github.com/solo-io/go-utils/errutils"
	"github.com/solo-io/solo-kit/pkg/api/v1/clients"
)

type ConfigSimpleEmitter interface {
	Snapshots(ctx context.Context) (<-chan *ConfigSnapshot, <-chan error, error)
}

func NewConfigSimpleEmitter(aggregatedWatch clients.ResourceWatch) ConfigSimpleEmitter {
	return NewConfigSimpleEmitterWithEmit(aggregatedWatch, make(chan struct{}))
}

func NewConfigSimpleEmitterWithEmit(aggregatedWatch clients.ResourceWatch, emit <-chan struct{}) ConfigSimpleEmitter {
	return &configSimpleEmitter{
		aggregatedWatch: aggregatedWatch,
		forceEmit:       emit,
	}
}

type configSimpleEmitter struct {
	forceEmit       <-chan struct{}
	aggregatedWatch clients.ResourceWatch
}

func (c *configSimpleEmitter) Snapshots(ctx context.Context) (<-chan *ConfigSnapshot, <-chan error, error) {
	snapshots := make(chan *ConfigSnapshot)
	errs := make(chan error)

	untyped, watchErrs, err := c.aggregatedWatch(ctx)
	if err != nil {
		return nil, nil, err
	}

	go errutils.AggregateErrs(ctx, errs, watchErrs, "config-emitter")

	go func() {
		originalSnapshot := ConfigSnapshot{}
		currentSnapshot := originalSnapshot.Clone()
		timer := time.NewTicker(time.Second * 1)
		sync := func() {
			if originalSnapshot.Hash() == currentSnapshot.Hash() {
				return
			}

			stats.Record(ctx, mConfigSnapshotOut.M(1))
			originalSnapshot = currentSnapshot.Clone()
			sentSnapshot := currentSnapshot.Clone()
			snapshots <- &sentSnapshot
		}

		defer func() {
			close(snapshots)
			close(errs)
		}()

		for {
			record := func() { stats.Record(ctx, mConfigSnapshotIn.M(1)) }

			select {
			case <-timer.C:
				sync()
			case <-ctx.Done():
				return
			case <-c.forceEmit:
				sentSnapshot := currentSnapshot.Clone()
				snapshots <- &sentSnapshot
			case untypedList := <-untyped:
				record()

				currentSnapshot = ConfigSnapshot{}
				for _, res := range untypedList {
					switch typed := res.(type) {
					case *Mesh:
						currentSnapshot.Meshes = append(currentSnapshot.Meshes, typed)
					case *MeshIngress:
						currentSnapshot.Meshingresses = append(currentSnapshot.Meshingresses, typed)
					case *MeshGroup:
						currentSnapshot.Meshgroups = append(currentSnapshot.Meshgroups, typed)
					case *RoutingRule:
						currentSnapshot.Routingrules = append(currentSnapshot.Routingrules, typed)
					case *SecurityRule:
						currentSnapshot.Securityrules = append(currentSnapshot.Securityrules, typed)
					case *TlsSecret:
						currentSnapshot.Tlssecrets = append(currentSnapshot.Tlssecrets, typed)
					case *gloo_solo_io.Upstream:
						currentSnapshot.Upstreams = append(currentSnapshot.Upstreams, typed)
					case *github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.Pod:
						currentSnapshot.Pods = append(currentSnapshot.Pods, typed)
					case *github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.Service:
						currentSnapshot.Services = append(currentSnapshot.Services, typed)
					default:
						select {
						case errs <- fmt.Errorf("ConfigSnapshotEmitter "+
							"cannot process resource %v of type %T", res.GetMetadata().Ref(), res):
						case <-ctx.Done():
							return
						}
					}
				}

			}
		}
	}()
	return snapshots, errs, nil
}