How to use emitOutput method of parallel_support Package

...38	//streaming endpoints39	mux.HandleFunc("/suite-will-begin", server.specSuiteWillBegin)40	mux.HandleFunc("/did-run", server.didRun)41	mux.HandleFunc("/suite-did-end", server.specSuiteDidEnd)42	mux.HandleFunc("/emit-output", server.emitOutput)43	//synchronization endpoints44	mux.HandleFunc("/before-suite-completed", server.handleBeforeSuiteCompleted)45	mux.HandleFunc("/before-suite-state", server.handleBeforeSuiteState)46	mux.HandleFunc("/have-nonprimary-procs-finished", server.handleHaveNonprimaryProcsFinished)47	mux.HandleFunc("/aggregated-nonprimary-procs-report", server.handleAggregatedNonprimaryProcsReport)48	mux.HandleFunc("/counter", server.handleCounter)49	mux.HandleFunc("/up", server.handleUp)50	mux.HandleFunc("/abort", server.handleAbort)51	go httpServer.Serve(server.listener)52}53//Stop the server54func (server *httpServer) Close() {55	server.listener.Close()56}57//The address the server can be reached it.  Pass this into the `ForwardingReporter`.58func (server *httpServer) Address() string {59	return "http://" + server.listener.Addr().String()60}61func (server *httpServer) GetSuiteDone() chan interface{} {62	return server.handler.done63}64func (server *httpServer) GetOutputDestination() io.Writer {65	return server.handler.outputDestination66}67func (server *httpServer) SetOutputDestination(w io.Writer) {68	server.handler.outputDestination = w69}70func (server *httpServer) RegisterAlive(node int, alive func() bool) {71	server.handler.registerAlive(node, alive)72}73//74// Streaming Endpoints75//76//The server will forward all received messages to Ginkgo reporters registered with `RegisterReporters`77func (server *httpServer) decode(writer http.ResponseWriter, request *http.Request, object interface{}) bool {78	defer request.Body.Close()79	if json.NewDecoder(request.Body).Decode(object) != nil {80		writer.WriteHeader(http.StatusBadRequest)81		return false82	}83	return true84}85func (server *httpServer) handleError(err error, writer http.ResponseWriter) bool {86	if err == nil {87		return false88	}89	switch err {90	case ErrorEarly:91		writer.WriteHeader(http.StatusTooEarly)92	case ErrorGone:93		writer.WriteHeader(http.StatusGone)94	case ErrorFailed:95		writer.WriteHeader(http.StatusFailedDependency)96	default:97		writer.WriteHeader(http.StatusInternalServerError)98	}99	return true100}101func (server *httpServer) specSuiteWillBegin(writer http.ResponseWriter, request *http.Request) {102	var report types.Report103	if !server.decode(writer, request, &report) {104		return105	}106	server.handleError(server.handler.SpecSuiteWillBegin(report, voidReceiver), writer)107}108func (server *httpServer) didRun(writer http.ResponseWriter, request *http.Request) {109	var report types.SpecReport110	if !server.decode(writer, request, &report) {111		return112	}113	server.handleError(server.handler.DidRun(report, voidReceiver), writer)114}115func (server *httpServer) specSuiteDidEnd(writer http.ResponseWriter, request *http.Request) {116	var report types.Report117	if !server.decode(writer, request, &report) {118		return119	}120	server.handleError(server.handler.SpecSuiteDidEnd(report, voidReceiver), writer)121}122func (server *httpServer) emitOutput(writer http.ResponseWriter, request *http.Request) {123	output, err := io.ReadAll(request.Body)124	if err != nil {125		writer.WriteHeader(http.StatusInternalServerError)126		return127	}128	var n int129	server.handleError(server.handler.EmitOutput(output, &n), writer)130}131func (server *httpServer) handleBeforeSuiteCompleted(writer http.ResponseWriter, request *http.Request) {132	var beforeSuiteState BeforeSuiteState133	if !server.decode(writer, request, &beforeSuiteState) {134		return135	}136	server.handleError(server.handler.BeforeSuiteCompleted(beforeSuiteState, voidReceiver), writer)...

1package parallel_support2import (3	"net/rpc"4	"time"5	"github.com/onsi/ginkgo/v2/types"6)7// TODO:8// - get RPC working9// - performance test10// - add DeferCleanup to test helper11type rpcClient struct {12	serverHost string13	client     *rpc.Client14}15func newRPCClient(serverHost string) *rpcClient {16	return &rpcClient{17		serverHost: serverHost,18	}19}20func (client *rpcClient) Connect() bool {21	var err error22	if client.client != nil {23		return true24	}25	client.client, err = rpc.DialHTTPPath("tcp", client.serverHost, "/")26	if err != nil {27		client.client = nil28		return false29	}30	return true31}32func (client *rpcClient) Close() error {33	return client.client.Close()34}35func (client *rpcClient) poll(method string, data interface{}) error {36	for {37		err := client.client.Call(method, voidSender, data)38		if err == nil {39			return nil40		}41		switch err.Error() {42		case ErrorEarly.Error():43			time.Sleep(POLLING_INTERVAL)44		case ErrorGone.Error():45			return ErrorGone46		case ErrorFailed.Error():47			return ErrorFailed48		default:49			return err50		}51	}52}53func (client *rpcClient) PostSuiteWillBegin(report types.Report) error {54	return client.client.Call("Server.SpecSuiteWillBegin", report, voidReceiver)55}56func (client *rpcClient) PostDidRun(report types.SpecReport) error {57	return client.client.Call("Server.DidRun", report, voidReceiver)58}59func (client *rpcClient) PostSuiteDidEnd(report types.Report) error {60	return client.client.Call("Server.SpecSuiteDidEnd", report, voidReceiver)61}62func (client *rpcClient) Write(p []byte) (int, error) {63	var n int64	err := client.client.Call("Server.EmitOutput", p, &n)65	return n, err66}67func (client *rpcClient) PostSynchronizedBeforeSuiteCompleted(state types.SpecState, data []byte) error {68	beforeSuiteState := BeforeSuiteState{69		State: state,70		Data:  data,71	}72	return client.client.Call("Server.BeforeSuiteCompleted", beforeSuiteState, voidReceiver)73}74func (client *rpcClient) BlockUntilSynchronizedBeforeSuiteData() (types.SpecState, []byte, error) {75	var beforeSuiteState BeforeSuiteState76	err := client.poll("Server.BeforeSuiteState", &beforeSuiteState)77	if err == ErrorGone {78		return types.SpecStateInvalid, nil, types.GinkgoErrors.SynchronizedBeforeSuiteDisappearedOnProc1()79	}80	return beforeSuiteState.State, beforeSuiteState.Data, err81}82func (client *rpcClient) BlockUntilNonprimaryProcsHaveFinished() error {83	return client.poll("Server.HaveNonprimaryProcsFinished", voidReceiver)84}85func (client *rpcClient) BlockUntilAggregatedNonprimaryProcsReport() (types.Report, error) {86	var report types.Report87	err := client.poll("Server.AggregatedNonprimaryProcsReport", &report)88	if err == ErrorGone {89		return types.Report{}, types.GinkgoErrors.AggregatedReportUnavailableDueToNodeDisappearing()90	}91	return report, err92}93func (client *rpcClient) FetchNextCounter() (int, error) {94	var counter int95	err := client.client.Call("Server.Counter", voidSender, &counter)96	return counter, err97}98func (client *rpcClient) PostAbort() error {99	return client.client.Call("Server.Abort", voidSender, voidReceiver)100}101func (client *rpcClient) ShouldAbort() bool {102	var shouldAbort bool103	client.client.Call("Server.ShouldAbort", voidSender, &shouldAbort)104	return shouldAbort105}...

1func main() {2    parallel_support.emitOutput(a)3}4func main() {5    parallel_support.emitOutput(a)6}7func main() {8    parallel_support.emitOutput(a)9}10func main() {11    parallel_support.emitOutput(a)12}13func main() {14    parallel_support.emitOutput(a)15}16func main() {17    parallel_support.emitOutput(a)18}19func main() {20    parallel_support.emitOutput(a)21}22func main() {23    parallel_support.emitOutput(a)24}25func main() {26    parallel_support.emitOutput(a)27}28func main() {29    parallel_support.emitOutput(a)30}

1import (2func main() {3	fmt.Println("Hello World!")4	parallel.EmitOutput(1, 1)5}6import (7type Output struct {8}9func EmitOutput(key int, value int) {10	fmt.Println("Key:", key, "Value:", value)11}12import (13func main() {14	fmt.Println("Hello World!")15	parallel.EmitOutput(1, 1)16}17import (18func main() {19	fmt.Println("Hello World!")20	parallel.EmitOutput(1, 1)21}

1import (2func main() {3	parallelSupport := parallel_support.NewParallelSupport()4	parallelSupport.Run(input, emitOutput, log)5}6func emitOutput(line string) {7	words := strings.Split(line, " ")8	for _, word := range words {9		parallel_support.Emit(word, "1")10	}11}12func reduce(key string, values []string) {13	for _, value := range values {14		intValue, _ := strconv.Atoi(value)15	}16	parallel_support.Emit(key, strconv.Itoa(count))17}18func log(key string, values []string) {19	for _, value := range values {20		fmt.Println(key + " " + value)21	}22}23import (

1import "fmt"2import "math/rand"3import "time"4import "runtime"5import "os"6import "strconv"7import "github.com/parallel_support"8func main() {9    numCores := runtime.NumCPU()10    runtime.GOMAXPROCS(numCores)11    numTasks, _ := strconv.Atoi(os.Args[1])12    tasks := make(chan int, numTasks)13    results := make(chan int, numTasks)14    output := make(chan string, numTasks)15    numTasksChannel := make(chan int, 1)16    numResultsChannel := make(chan int, 1)17    for i := 0; i < numTasks; i++ {18    }19    close(tasks)20    parallel_support := parallel_support.parallel_support{}21    parallel_support.StartParallelTasks(tasks, results, output, numTasksChannel, numResultsChannel)22    close(results)23    close(output)24    close(numTasksChannel)25    close(numResultsChannel)26    fmt.Println("Number of results: ", numResults)27    for result := range output {28        fmt.Println(result)29    }30}31import "fmt"32import "math/rand"33import "time"34import "runtime"35import "os"36import "strconv"37import "github.com/parallel_support"38func main() {39    numCores := runtime.NumCPU()