<commit_before>// Broadcast over channels.
package bcast

/*
   bcast package for Go. Broadcasting on a set of channels.

   Copyright © 2013 <NAME> <<EMAIL>>.
   All rights reserved.
   Use of this source code is governed by a BSD-style
   license that can be found in the LICENSE file.
*/

import (
	"sync"
	"time"
	//"fmt"
)

// Internal structure to pack messages together with info about sender.
type Message struct {
	sender  chan interface{}
	payload interface{}
}

// Represents member of broadcast group.
type Member struct {
	group *Group           // send messages to others directly to group.In
	In    chan interface{} // (public) get messages from others to own channel
	//NonReceived int
}

// Represents broadcast group.
type Group struct {
	l     sync.Mutex
	in    chan Message       // receive broadcasts from members
	out   []chan interface{} // broadcast messages to members
	count int
	close chan bool
}

// Create new broadcast group.
func NewGroup() *Group {
	in := make(chan Message)
	close := make(chan bool)
	return &Group{in: in, close: close}
}

func (r *Group) MemberCount() int {
	return r.count
}

func (r *Group) Members() []chan interface{} {
	r.l.Lock()
	res := r.out[:]
	r.count = len(r.out)
	r.l.Unlock()
	return res
}

func (r *Group) Add(out chan interface{}) {
	r.l.Lock()
	r.out = append(r.out, out)
	r.count = len(r.out)
	r.l.Unlock()
	return
}

func (r *Group) Remove(received Message) {
	r.l.Lock()
	for i, addr := range r.out {
		if addr == received.payload.(Member).In && received.sender == received.payload.(Member).In {
			r.out = append(r.out[:i], r.out[i+1:]...)
			r.count = len(r.out)
			break
		}
	}
	r.l.Unlock()
	return
}

// Close the group immediately
func (r *Group) Close() {
	r.close <- true
}

// Broadcast messages received from one group member to others.
// If incoming messages not arrived during `timeout` then function returns.
// Set `timeout` to zero to set unlimited timeout or use Broadcasting().
func (r *Group) Broadcasting(timeout time.Duration) {
	for {
		select {
		case received := <-r.in:
			switch received.payload.(type) {
			default: // receive a payload and broadcast it
				for _, member := range r.Members() {
					if received.sender != member { // not return broadcast to sender

						/*
							select {
							case member <- received.payload:
								fmt.Println("sent message", received.payload)
							default:
								fmt.Println("no message sent")
							}
						*/
						go func(out chan interface{}, received *Message) { // non blocking
							out <- received.payload
						}(member, &received)

					}
				}
			}
		case <-time.After(timeout):
			if timeout > 0 {
				return
			}
		case <-r.close:
			return
		}
	}
}

// Broadcast messages received from one group member to others.
// See https://github.com/NimbleIndustry/bcast/issues/4 for rationale
func (r *Group) ContinuousBroadcasting() {
	for {
		select {
		case <-r.close:
			return
		case received := <-r.in:
			switch received.payload.(type) {
			default: // receive a payload and broadcast it
				for _, member := range r.Members() {
					if received.sender != member { // not return broadcast to sender

						/*<commit_msg>Adapt functions and repository names<commit_after>// Broadcast over channels.
package bcast

/*
   bcast package for Go. Broadcasting on a set of channels.

   Copyright © 2013 <NAME> <<EMAIL>>.
   All rights reserved.
   Use of this source code is governed by a BSD-style
   license that can be found in the LICENSE file.
*/

import (
	"sync"
	"time"
	//"fmt"
)

// Internal structure to pack messages together with info about sender.
type Message struct {
	sender  chan interface{}
	payload interface{}
}

// Represents member of broadcast group.
type Member struct {
	group *Group           // send messages to others directly to group.In
	In    chan interface{} // (public) get messages from others to own channel
	//NonReceived int
}

// Represents broadcast group.
type Group struct {
	l     sync.Mutex
	in    chan Message       // receive broadcasts from members
	out   []chan interface{} // broadcast messages to members
	count int
	close chan bool
}

// Create new broadcast group.
func NewGroup() *Group {
	in := make(chan Message)
	close := make(chan bool)
	return &Group{in: in, close: close}
}

func (r *Group) MemberCount() int {
	return r.count
}

func (r *Group) Members() []chan interface{} {
	r.l.Lock()
	res := r.out[:]
	r.count = len(r.out)
	r.l.Unlock()
	return res
}

func (r *Group) Add(out chan interface{}) {
	r.l.Lock()
	r.out = append(r.out, out)
	r.count = len(r.out)
	r.l.Unlock()
	return
}

func (r *Group) Remove(received Message) {
	r.l.Lock()
	for i, addr := range r.out {
		if addr == received.payload.(Member).In && received.sender == received.payload.(Member).In {
			r.out = append(r.out[:i], r.out[i+1:]...)
			r.count = len(r.out)
			break
		}
	}
	r.l.Unlock()
	return
}

// Close the group immediately
func (r *Group) Close() {
	r.close <- true
}

// Obsoleted by BroadcastFor()
func (r *Group) Broadcasting(timeout time.Duration) {
	r.BroadcastFor(timeout)
}

// Broadcast messages received from one group member to others.
// If incoming messages not arrived during `timeout` then function returns.
func (r *Group) BroadcastFor(timeout time.Duration) {
	if timeout == 0 {
		r.Broadcast()
		return
	}
	for {
		select {
		case received := <-r.in:
			switch received.payload.(type) {
			default: // receive a payload and broadcast it
				for _, member := range r.Members() {
					if received.sender != member { // not return broadcast to sender

						/*
							select {
							case member <- received.payload:
								fmt.Println("sent message", received.payload)
							default:
								fmt.Println("no message sent")
							}
						*/
						go func(out chan interface{}, received *Message) { // non blocking
							out <- received.payload
						}(member, &received)

					}
				}
			}
		case <-time.After(timeout):
			if timeout > 0 {
				return
			}
		case <-r.close:
			return
		}
	}
}

// Broadcast messages received from one group member to others.
// See https://github.com/grafov/bcast/issues/4 for rationale.
func (r *Group) Broadcast() {
	for {
		select {
		case <-r.close:
			return
		case received := <-r.in:
			switch received.payload.(type) {
			default: // receive a payload and broadcast it
				for _, member := range r.Members() {
					if received.sender != member { // not return broadcast to sender

						/*
{'fraction_non_alphanumeric': 0.12631905648665426, 'fraction_numerical': 0.0023277467411545625, 'mean_word_length': 2.7646028037383177, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 11, 'https://': 2, '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': '6084424', 'n_tokens_mistral': 2321, 'n_tokens_neox': 1999, 'n_words': 703}