<commit_before>package capacitor

import (
	"sync"
	"sync/atomic"
	"time"
)

type Interface interface {
	Submit(func())
	Discharge()
	Cancel()
}

func New() Interface {
	c := &capacitor{
		delay: time.Second,
	}

	return c
}

type delayer struct {
	current   atomic.Value
	terminate chan bool
}

func (d *delayer) cancel(discharge bool) {
	d.terminate <- discharge
}

func (d *delayer) execute() {
	if f, ok := d.current.Load().(func()); f != nil && ok {
		f()
	}
}

func (d *delayer) run(timer <-chan time.Time, stop func() bool) {
	defer stop()
	for {
		select {
		case <-timer:
			select {
			case discharge := <-d.terminate:
				if discharge {
					d.execute()
				}

			default:
				d.execute()
			}

			return

		case discharge := <-d.terminate:
			if discharge {
				d.execute()
			}
		}
	}
}

type capacitor struct {
	lock  sync.Mutex
	delay time.Duration
	d     *delayer
}

func (c *capacitor) Submit(v func()) {
	c.lock.Lock()
	if c.d == nil {
		c.d = &delayer{
			terminate: make(chan bool, 1),
		}

		timer := time.NewTimer(c.delay)
		go c.d.run(timer.C, timer.Stop)
	}

	c.d.current.Store(v)
	c.lock.Unlock()
}

func (c *capacitor) Discharge() {
	c.lock.Lock()
	if c.d != nil {
		c.d.cancel(true)
		c.d = nil
	}

	c.lock.Unlock()
}

func (c *capacitor) Cancel() {
	c.lock.Lock()
	if c.d != nil {
		c.d.cancel(false)
		c.d = nil
	}

	c.lock.Unlock()
}<commit_msg>Fleshed out the atomic operations<commit_after>package capacitor

import (
	"sync"
	"sync/atomic"
	"time"

	"github.com/Comcast/webpa-common/clock"
)

type Interface interface {
	Submit(func())
	Discharge()
	Cancel()
}

func New() Interface {
	c := &capacitor{
		delay: time.Second,
		c:     clock.System(),
	}

	return c
}

type delayer struct {
	current   atomic.Value
	timer     clock.Timer
	terminate chan bool
	reset     func()
}

func (d *delayer) discharge() {
	d.terminate <- true
}

func (d *delayer) cancel() {
	d.terminate <- false
}

func (d *delayer) execute() {
	if f, ok := d.current.Load().(func()); f != nil && ok {
		f()
	}
}

func (d *delayer) run() {
	defer d.timer.Stop()
	defer d.reset()

	for {
		select {
		case <-d.timer.C():
			select {
			case discharge := <-d.terminate:
				if discharge {
					d.execute()
				}

			default:
				d.execute()
			}

			return

		case discharge := <-d.terminate:
			if discharge {
				d.execute()
			}
		}
	}
}

type capacitor struct {
	lock  sync.Mutex
	delay time.Duration
	c     clock.Interface
	d     *delayer
}

// reset produces a closure that a given delayer must call to clean up the enclosing capacitor.
// The returned closure atomically sets the delayer to nil if and only if it matched the given delayer.
// This allows for barging between the public interface methods.
func (c *capacitor) reset(d *delayer) func() {
	return func() {
		c.lock.Lock()
		if c.d == d {
			c.d = nil
		}
		c.lock.Unlock()
	}
}

func (c *capacitor) Submit(v func()) {
	c.lock.Lock()
	if c.d == nil {
		c.d = &delayer{
			terminate: make(chan bool, 1),
			timer:     c.c.NewTimer(c.delay),
		}

		c.d.current.Store(v)
		c.d.reset = c.reset(c.d)
		go c.d.run()
	} else {
		c.d.current.Store(v)
	}

	c.lock.Unlock()
}

func (c *capacitor) Discharge() {
	c.lock.Lock()
	if c.d != nil {
		c.d.discharge()
		c.d = nil
	}

	c.lock.Unlock()
}

func (c *capacitor) Cancel() {
	c.lock.Lock()
	if c.d != nil {
		c.d.cancel()
		c.d = nil
	}

	c.lock.Unlock()
}
{'fraction_non_alphanumeric': 0.1610320284697509, 'fraction_numerical': 0.0005931198102016608, 'mean_word_length': 2.654387865655471, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5072723', 'n_tokens_mistral': 1512, 'n_tokens_neox': 1353, 'n_words': 344}