<commit_before>// Copyright 2015 <NAME>. All rights reserved.
// Use of this source code is governed by the MIT license that can be found in the LICENSE file.

package pubsub

import (
	"sync"
)

//TODO: optional high water mark
// - sequence number on message
// - atomic head sequence number on topic
// - subscription sender checks current-to-head delay against the HWM
// - discard messages

// Topic
type Topic interface {
	// Publish publishes a new message to all subscribers.
	Publish(msg interface{})

	NumSubscribers() int

	Subscribe() Subscription
}

type Subscription interface {
	C() <-chan interface{}

	// Unsubscribe frees the resources which are associated with this Subscription.
	// Consumers of this API must call Unsubscribe.
	Unsubscribe()
}

func NewTopic() Topic { return &ct{head: newm()} }

type ct struct {
	mu     sync.Mutex
	head   *m
	numSub int
}
type cs struct {
	t       *ct
	current *m
	recv    chan interface{}
	stop    chan struct{}
}
type m struct {
	ready chan struct{} // guards the fields below, a closed ready channel is indicative of message readiness
	msg   interface{}   // the published message
	next  *m            // the next message holder object in the chain of published messages
}

func newm() *m { return &m{ready: make(chan struct{})} }

var _ Topic = (*ct)(nil)
var _ Subscription = (*cs)(nil)

func (t *ct) Publish(msg interface{}) {
	next := newm()

	t.mu.Lock()
	defer t.mu.Unlock()

	if t.numSub == 0 {
		return
	}

	t.head.msg = msg
	t.head.next = next
	close(t.head.ready)
	t.head = next
}

func (t *ct) NumSubscribers() int {
	t.mu.Lock()
	defer t.mu.Unlock()

	return t.numSub
}

func (t *ct) Subscribe() Subscription {
	t.mu.Lock()
	head := t.head
	t.numSub++
	t.mu.Unlock()

	sub := &cs{
		t:       t,
		current: head,
		recv:    make(chan interface{}),
		stop:    make(chan struct{}),
	}
	go sub.sender()
	return sub
}

func (s *cs) C() <-chan interface{} { return s.recv }
func (s *cs) Unsubscribe() {
	s.t.mu.Lock()
	s.t.numSub--
	s.t.mu.Unlock()
	close(s.stop)
}

func (s *cs) sender() {
	defer func() {
		s.current = nil
		close(s.recv)
	}()

	var msg interface{}
	for {
		if msg == nil {
			// receive next message or stop
			select {
			case <-s.stop:
				return
			case <-s.current.ready:
				msg = s.current.msg
				s.current = s.current.next
			}
		} else {
			// send message or stop
			select {
			case <-s.stop:
				return
			case s.recv <- msg:
				msg = nil
			}
		}
	}
}<commit_msg>comments<commit_after>// Copyright 2015 <NAME>. All rights reserved.
// Use of this source code is governed by the MIT license that can be found in the LICENSE file.

package pubsub

import (
	"sync"
)

//TODO: optional high water mark
// - sequence number on message
// - atomic head sequence number on topic
// - subscription sender checks current-to-head delay against the HWM
// - discard messages

// Topic
type Topic interface {
	// Publish publishes a new message to all subscribers.
	Publish(msg interface{})

	// Subscribe creates a new Subscription on which newly published messages can be received.
	Subscribe() Subscription

	// NumSubscribers returns the current number of subscribers.
	NumSubscribers() int
}

type Subscription interface {
	// C returns the receive-only channel on which newly published messages are received.
	C() <-chan interface{}

	// Unsubscribe frees the resources which are associated with this Subscription.
	// Consumers must call Unsubscribe to prevent memory/goroutine leaks.
	Unsubscribe()
}

func NewTopic() Topic { return &ct{head: newm()} }

type ct struct {
	mu     sync.Mutex
	head   *m
	numSub int
}
type cs struct {
	t       *ct
	current *m
	recv    chan interface{}
	stop    chan struct{}
}
type m struct {
	ready chan struct{} // guards the fields below, a closed ready channel is indicative of message readiness
	msg   interface{}   // the published message
	next  *m            // the next message holder object in the chain of published messages
}

func newm() *m { return &m{ready: make(chan struct{})} }

var _ Topic = (*ct)(nil)
var _ Subscription = (*cs)(nil)

func (t *ct) Publish(msg interface{}) {
	next := newm()

	t.mu.Lock()
	defer t.mu.Unlock()

	if t.numSub == 0 {
		return
	}

	t.head.msg = msg
	t.head.next = next
	close(t.head.ready)
	t.head = next
}

func (t *ct) NumSubscribers() int {
	t.mu.Lock()
	defer t.mu.Unlock()

	return t.numSub
}

func (t *ct) Subscribe() Subscription {
	t.mu.Lock()
	head := t.head
	t.numSub++
	t.mu.Unlock()

	sub := &cs{
		t:       t,
		current: head,
		recv:    make(chan interface{}),
		stop:    make(chan struct{}),
	}
	go sub.subscriberLoop()
	return sub
}

func (s *cs) C() <-chan interface{} { return s.recv }

func (s *cs) Unsubscribe() {
	s.t.mu.Lock()
	s.t.numSub--
	s.t.mu.Unlock()
	close(s.stop)
}

func (s *cs) subscriberLoop() {
	defer func() {
		s.current = nil
		close(s.recv)
	}()

	var msg interface{}
	for {
		if msg == nil {
			// receive next message or stop
			select {
			case <-s.stop:
				return
			case <-s.current.ready:
				msg = s.current.msg
				s.current = s.current.next
			}
		} else {
			// send message or stop
			select {
			case <-s.stop:
				return
			case s.recv <- msg:
				msg = nil
			}
		}
	}
}
{'fraction_non_alphanumeric': 0.1255527783118631, 'fraction_numerical': 0.0019227071716977505, 'mean_word_length': 2.813782991202346, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 5, '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': '179438', 'n_tokens_mistral': 1914, 'n_tokens_neox': 1765, 'n_words': 619}