<commit_before>package backends

import (
	"io"
	"fmt"
	"github.com/docker/libswarm/beam"
	beamutils "github.com/docker/libswarm/beam/utils"
	"strings"
	"sync"
	"time"
)

// New returns a new engine, with all backends
// registered but not activated.
// To activate a backend, call a job on the resulting
// engine, named after the desired backend.
//
// Example: `New().Job("debug").Run()`
func New() beam.Sender {
	backends := beamutils.NewHub()
	backends.RegisterName("cd", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		return false, fmt.Errorf("no such backend: %s\n", strings.Join(msg.Args, " "))
	})
	backends.RegisterName("cd", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		if len(msg.Args) > 0 && msg.Args[0] == "debug" {
			debugr, debugw, err := out.Send(&beam.Message{Name: "register"}, beam.R|beam.W)
			if err != nil {
				return false, err
			}
			go func() {
				for {
					msg, msgr, msgw, err := debugr.Receive(beam.R | beam.W)
					if err == io.EOF {
						return
					}
					if err != nil {
						return
					}
					fmt.Printf("[DEBUG] %s %s\n", msg.Name, strings.Join(msg.Args, " "))
					// FIXME: goroutine?
					Splice(debugw, msg, msgr, msgw)
				}
			}()
			return false, nil
		}
		return true, nil
	})
	backends.RegisterName("fakeclient", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		backends.RegisterTask(func(r beam.Receiver, w beam.Sender) error {
			for {
				time.Sleep(1 * time.Second)
				_, _, err := w.Send(&beam.Message{Name: "log", Args: []string{"fake client reporting for duty"}}, 0)
				if err != nil {
					return err
				}
				containersR, _, err := w.Send(&beam.Message{Name: "containers"}, beam.R)
				if err != nil {
					return err
				}
				go beamutils.Copy(beam.NopSender{}, containersR)
			}
		})
		return true, nil
	})
	return backends
}

func Splice(dst beam.Sender, msg *beam.Message, r beam.Receiver, w beam.Sender) error {
	dstR, dstW, err := dst.Send(msg, beam.R|beam.W)
	if err != nil {
		return err
	}
	defer dstW.Close()
	var tasks sync.WaitGroup
	_copy := func(dst beam.Sender, src beam.Receiver) {
		tasks.Add(1)
		go func() {
			defer tasks.Done()
			beamutils.Copy(dst, src)
			dst.Close()
		}()<commit_msg>fakeclient backend: switch to object model navigation<commit_after>package backends

import (
	"io"
	"fmt"
	"github.com/docker/libswarm/beam"
	beamutils "github.com/docker/libswarm/beam/utils"
	"strings"
	"sync"
	"time"
)

// New returns a new engine, with all backends
// registered but not activated.
// To activate a backend, call a job on the resulting
// engine, named after the desired backend.
//
// Example: `New().Job("debug").Run()`
func New() beam.Sender {
	backends := beamutils.NewHub()
	backends.RegisterName("cd", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		return false, fmt.Errorf("no such backend: %s\n", strings.Join(msg.Args, " "))
	})
	backends.RegisterName("cd", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		if len(msg.Args) > 0 && msg.Args[0] == "debug" {
			debugr, debugw, err := out.Send(&beam.Message{Name: "register"}, beam.R|beam.W)
			if err != nil {
				return false, err
			}
			go func() {
				for {
					msg, msgr, msgw, err := debugr.Receive(beam.R | beam.W)
					if err == io.EOF {
						return
					}
					if err != nil {
						return
					}
					fmt.Printf("[DEBUG] %s %s\n", msg.Name, strings.Join(msg.Args, " "))
					// FIXME: goroutine?
					Splice(debugw, msg, msgr, msgw)
				}
			}()
			return false, nil
		}
		return true, nil
	})
	backends.RegisterName("cd", func(msg *beam.Message, in beam.Receiver, out beam.Sender, next beam.Sender) (bool, error) {
		if len(msg.Args) > 0 && msg.Args[0] == "fakeclient" {
			_, fakew, err := out.Send(&beam.Message{Name: "register"}, beam.R|beam.W)
			if err != nil {
				return false, err
			}
			go func() {
				defer fmt.Printf("[FAKECLIENT] done\n")
				for {
					time.Sleep(1 * time.Second)
					fmt.Printf("[FAKECLIENT] heartbeat\n")
					_, _, err := fakew.Send(&beam.Message{Name: "log", Args: []string{"fake client reporting for duty"}}, 0)
					if err != nil {
						return
					}
					containersR, _, err := fakew.Send(&beam.Message{Name: "containers"}, beam.R)
					if err != nil {
						return
					}
					go beamutils.Copy(beam.NopSender{}, containersR)
				}
			}()
			return false, nil
		}
		return true, nil
	})
	return backends
}

func Splice(dst beam.Sender, msg *beam.Message, r beam.Receiver, w beam.Sender) error {
	dstR, dstW, err := dst.Send(msg, beam.R|beam.W)
	if err != nil {
		return err
	}
	defer dstW.Close()
	var tasks sync.WaitGroup
	_copy := func(dst beam.Sender, src beam.Receiver) {
		tasks.Add(1)
		go func() {
			defer tasks.Done()
			beamutils.Copy(dst, src)
			dst.Close()
		}()
{'fraction_non_alphanumeric': 0.16260162601626016, 'fraction_numerical': 0.0024390243902439024, 'mean_word_length': 3.094009983361065, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, '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': '6232711', 'n_tokens_mistral': 2066, 'n_tokens_neox': 1765, 'n_words': 457}