ronantakizawa/codereview-bench huggingface dataset:
Repository: gogs/gogs
File: internal/lfsutil/storage.go (Go)

Code before:
		return 0, ErrInvalidOID
	}

	fpath := s.storagePath(oid)
	dir := filepath.Dir(fpath)

	defer rc.Close()

	if err := os.MkdirAll(dir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create directories")
	}

	// If the object file already exists, skip the upload and return the
	// existing file's size.
	if fi, err := os.Stat(fpath); err == nil {
		_, _ = io.Copy(io.Discard, rc)
		return fi.Size(), nil
	}

	// Write to a temp file and verify the content hash before publishing.
	// This ensures the final path always contains a complete, hash-verified
	// file, even when concurrent uploads of the same OID race.
	if err := os.MkdirAll(s.TempDir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create temp directory")
	}
	tmp, err := os.CreateTemp(s.TempDir, ".lfs-upload-*")
	if err != nil {
		return 0, errors.Wrap(err, "create temp file")
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)

	hash := sha256.New()
	written, err := io.Copy(tmp, io.TeeReader(rc, hash))
	if closeErr := tmp.Close(); err == nil && closeErr != nil {
		err = closeErr
	}
	if err != nil {
		return 0, errors.Wrap(err, "write object file")
	}

	if computed := hex.EncodeToString(hash.Sum(nil)); computed != string(oid) {
		return 0, ErrOIDMismatch
	}

	if err := os.Rename(tmpPath, fpath); err != nil {
		return 0, errors.Wrap(err, "publish object file")
	}
	return written, nil
}


Review comment (suggestion):
```suggestion
	tmp, err := os.CreateTemp(s.TempDir, "upload-*")
```

Code after:
		return 0, ErrInvalidOID
	}

	fpath := s.storagePath(oid)
	dir := filepath.Dir(fpath)

	defer rc.Close()

	if err := os.MkdirAll(dir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create directories")
	}

	// If the object file already exists, skip the upload and return the
	// existing file's size.
	if fi, err := os.Stat(fpath); err == nil {
		_, _ = io.Copy(io.Discard, rc)
		return fi.Size(), nil
	}

	// Write to a temp file and verify the content hash before publishing.
	// This ensures the final path always contains a complete, hash-verified
	// file, even when concurrent uploads of the same OID race.
	if err := os.MkdirAll(s.TempDir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create temp directory")
	}
	tmp, err := os.CreateTemp(s.TempDir, "upload-*")
	if err != nil {
		return 0, errors.Wrap(err, "create temp file")
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)

	hash := sha256.New()
	written, err := io.Copy(tmp, io.TeeReader(rc, hash))
	if closeErr := tmp.Close(); err == nil && closeErr != nil {
		err = closeErr
	}
	if err != nil {
		return 0, errors.Wrap(err, "write object file")
	}

	if computed := hex.EncodeToString(hash.Sum(nil)); computed != string(oid) {
		return 0, ErrOIDMismatch
	}

	if err := os.Rename(tmpPath, fpath); err != nil && !os.IsExist(err) {
		return 0, errors.Wrap(err, "publish object file")
	}
	return written, nil
}
Repository: gogs/gogs
File: internal/lfsutil/storage.go (Go)

Code before:
	// file, even when concurrent uploads of the same OID race.
	if err := os.MkdirAll(s.TempDir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create temp directory")
	}
	tmp, err := os.CreateTemp(s.TempDir, "upload-*")
	if err != nil {
		return 0, errors.Wrap(err, "create temp file")
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)

	hash := sha256.New()
	written, err := io.Copy(tmp, io.TeeReader(rc, hash))
	if closeErr := tmp.Close(); err == nil && closeErr != nil {
		err = closeErr
	}
	if err != nil {
		return 0, errors.Wrap(err, "write object file")
	}

	if computed := hex.EncodeToString(hash.Sum(nil)); computed != string(oid) {
		return 0, ErrOIDMismatch
	}

	if err := os.Rename(tmpPath, fpath); err != nil {
		return 0, errors.Wrap(err, "publish object file")
	}
	return written, nil
}

func (s *LocalStorage) Download(oid OID, w io.Writer) error {
	fpath := s.storagePath(oid)
	if !osutil.IsFile(fpath) {
		return ErrObjectNotExist
	}

	r, err := os.Open(fpath)
	if err != nil {
		return errors.Wrap(err, "open file")
	}
	defer r.Close()

	_, err = io.Copy(w, r)
	if err != nil {
		return errors.Wrap(err, "copy file")
	}
	return nil
}


Review comment (bug):
Is there a specific error type we can assert so if the error is about file already exist, we don't need to fail the client request.

Code after:
	// file, even when concurrent uploads of the same OID race.
	if err := os.MkdirAll(s.TempDir, os.ModePerm); err != nil {
		return 0, errors.Wrap(err, "create temp directory")
	}
	tmp, err := os.CreateTemp(s.TempDir, "upload-*")
	if err != nil {
		return 0, errors.Wrap(err, "create temp file")
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)

	hash := sha256.New()
	written, err := io.Copy(tmp, io.TeeReader(rc, hash))
	if closeErr := tmp.Close(); err == nil && closeErr != nil {
		err = closeErr
	}
	if err != nil {
		return 0, errors.Wrap(err, "write object file")
	}

	if computed := hex.EncodeToString(hash.Sum(nil)); computed != string(oid) {
		return 0, ErrOIDMismatch
	}

	if err := os.Rename(tmpPath, fpath); err != nil && !os.IsExist(err) {
		return 0, errors.Wrap(err, "publish object file")
	}
	return written, nil
}

func (s *LocalStorage) Download(oid OID, w io.Writer) error {
	fpath := s.storagePath(oid)
	if !osutil.IsFile(fpath) {
		return ErrObjectNotExist
	}

	r, err := os.Open(fpath)
	if err != nil {
		return errors.Wrap(err, "open file")
	}
	defer r.Close()

	_, err = io.Copy(w, r)
	if err != nil {
		return errors.Wrap(err, "copy file")
	}
	return nil
}
Repository: google/adk-go
File: tool.go (Go)

Code before:
}

type ToolArgs[TArgs any] struct {
	Args        TArgs        `json:"arguments,omitempty"`
	ToolContext *ToolContext `json:"-"`
}

type ToolResults[TResults any] struct {
	Results TResults `json:"results,omitempty"`
}

// TODO: Implement an MCP client as a tool to validate whether
// a concrete type is sufficient for tool use.

// TODO: Validate the Tool representation for long running tools.
// Introduce LongRunningTool as an additional type if needed.

type Tool struct {
	Name        string
	Description string
	InputSchema *jsonschema.Schema
}

type Handler[TArgs, TResults any] func(context.Context, *ToolArgs[TArgs]) (*ToolResults[TResults], error)

func NewTool[TArgs, TResults any](name string, desc string, handler Handler[TArgs, TResults]) *Tool {
	panic("not implemented")
}


Review comment (suggestion):
`Tool` has no field for the handler, so this won't work as written.
In our MCP implementation we define `ServerTool` for this purpose.


Code after:
)

// ToolContext is the tool invocation context.
type ToolContext struct {
	// The invocation context of the tool call.
	InvocationContext *InvocationContext

	// The function call id of the current tool call.
	// This id was returned in the function call event from LLM to identify
	// a function call. If LLM didn't return an id, ADK will assign one t it.
	// This id is used to map function call response to the original function call.
	FunctionCallID string

	// The event actions of the current tool call.
	EventActions *EventActions
}

// TODO: Implement an MCP client as a tool to validate whether
// a concrete type is sufficient for tool use.

// Tool is an ADK tool.
type Tool struct {
	Name        string
	Description string
	InputSchema *jsonschema.Schema
	// TODO: OutputSchema

	rawHandler rawToolHandler
}

// Run executes the tool with the provided context and yields events.
func (t *Tool) Run(ctx context.Context, tc *ToolContext, args map[string]any) (map[string]any, error) {
	return t.rawHandler(ctx, tc, args)
}

// ToolHandler is the execution handler of a tool.
type ToolHandler[In, Out any] func(ctx context.Context, input In) (output Out, err error)

type rawToolHandler func(ctx context.Context, tc *ToolContext, args map[string]any) (map[string]any, error)

// NewTool creates a new tool with a name, description, and the provided handler.
// Input schema is automatically inferred from the input and output types.
func NewTool[In, Out any](name string, description string, handler ToolHandler[In, Out]) *Tool {
	// TODO(jbd): Add ToolOption as variadic arguments to NewTool.
	ischema, err := jsonschema.For[In]()
	if err != nil {
		panic(fmt.Errorf("NewTool(%q): %w", name, err))
	}
	rawHandler := func(ctx context.Context, tc *ToolContext, args map[string]any) (map[string]any, error) {
		panic("not yet implemented")
		// TODO: Handle function call request from tc.InvocationContext.
Repository: google/adk-go
File: agent/utils.go (Go)

Code before:
	}
	for _, p := range c.Parts {
		if p.FunctionCall != nil {
			ret = append(ret, p.FunctionCall)
		}
	}
	return
}

// functionResponses extracts all FunctionResponse parts from the content.
func functionResponses(c *genai.Content) (ret []*genai.FunctionResponse) {
	if c == nil {
		return nil
	}
	for _, p := range c.Parts {
		if p.FunctionResponse != nil {
			ret = append(ret, p.FunctionResponse)
		}
	}
	return ret
}

// content is a convenience function that returns the genai.Content
// in the event.
func content(ev *adk.Event) *genai.Content {
	if ev == nil || ev.LLMResponse == nil || ev.LLMResponse.Content == nil {
		return nil
	}
	return ev.LLMResponse.Content
}


Review comment (suggestion):
```suggestion
	if ev == nil || ev.LLMResponse == nil {
```

Code after:
	}
	for _, p := range c.Parts {
		if p.FunctionCall != nil {
			ret = append(ret, p.FunctionCall)
		}
	}
	return ret
}

// functionResponses extracts all FunctionResponse parts from the content.
func functionResponses(c *genai.Content) (ret []*genai.FunctionResponse) {
	if c == nil {
		return nil
	}
	for _, p := range c.Parts {
		if p.FunctionResponse != nil {
			ret = append(ret, p.FunctionResponse)
		}
	}
	return ret
}

// content is a convenience function that returns the genai.Content
// in the event.
func content(ev *adk.Event) *genai.Content {
	if ev == nil || ev.LLMResponse == nil {
		return nil
	}
	return ev.LLMResponse.Content
}
Repository: google/adk-go
File: agent/base_agent.go (Go)

Code before:

	Self adk.Agent
}

func NewAgentBase(name string, opts ...Option) *AgentBase {
	b := &AgentBase{name: name}
	for _, opt := range opts {
		opt.apply2Base(b)
	}
	return b
}

var _ adk.Agent = (*AgentBase)(nil)

func (a *AgentBase) Name() string           { return a.name }
func (a *AgentBase) Description() string    { return a.description }
func (a *AgentBase) Parent() adk.Agent      { return a.parentAgent }
func (a *AgentBase) SubAgents() []adk.Agent { return a.subAgents }
func (a *AgentBase) Run(ctx context.Context, parentCtx *adk.InvocationContext) iter.Seq2[*adk.Event, error] {
	panic("unimplemented")
}

func (a *AgentBase) _base_() *AgentBase { return a }

// AddSubAgents adds the agents to the subagent list.
func (a *AgentBase) AddSubAgents(agents ...adk.Agent) {
	for _, subagent := range agents {
		a.subAgents = append(a.subAgents, subagent)

		if s, ok := subagent.(interface{ _base_() *AgentBase }); ok {
			if base := s._base_(); base != nil {
				base.parentAgent = a.Self
			}
		}
	}
}


Review comment (suggestion):
WDYT if we make an Option -- `WithSubAgents`? This will be then consistent with the rest of the setup, like `WithDescription` etc.

Code after:
	   agent := &MyCustomAgent{}
	   base, err := agent.NewBaseAgent(name, agent, opts...)}
	   if err != nil { return nil, err }
	   agent.BaseAgent = base
	    ...
	   return agent, nil
	}
*/
type BaseAgent struct {
	name        string
	description string
	parentAgent adk.Agent
	subAgents   []adk.Agent

	self adk.Agent
}

// NewBaseAgent returns a BaseAgent that can be the base of the implementation agent.
func NewBaseAgent(name string, implementation adk.Agent, opts ...Option) (*BaseAgent, error) {
	if implementation == nil {
		return nil, fmt.Errorf("implementation is nil")
	}
	b := &BaseAgent{name: name, self: implementation}
	for _, opt := range opts {
		if err := opt.apply2Base(b); err != nil {
			return nil, err
		}
	}
	return b, nil
}

func (a *BaseAgent) Name() string           { return a.name }
func (a *BaseAgent) Description() string    { return a.description }
func (a *BaseAgent) Parent() adk.Agent      { return a.parentAgent }
func (a *BaseAgent) SubAgents() []adk.Agent { return a.subAgents }

// TODO: Should we export it as Base() and include it in the interface?
// That will allows custom agents to wrap its BaseAgent instead of embedding.
func (a *BaseAgent) _base_() *BaseAgent { return a }

func asBaseAgent(a adk.Agent) *BaseAgent {
	if b, ok := a.(interface{ _base_() *BaseAgent }); ok {
		return b._base_()
	}
	return nil
}

// AddSubAgents adds the agents to the subagent list.
func (a *BaseAgent) AddSubAgents(agents ...adk.Agent) error {
	names := map[string]bool{}
	for _, subagent := range a.subAgents {
Repository: google/adk-go
File: agent/base_agent.go (Go)

Code before:
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package agent

import (
	"context"
	"iter"

	"github.com/google/adk-go"
)

// Option configures the Base agent.
type Option interface {
	apply2Base(*AgentBase)
}

type optionFunc func(*AgentBase)

func (o optionFunc) apply2Base(b *AgentBase) { o(b) }

// WithName sets the agent name.
func WithName(name string) Option {
	return optionFunc(func(b *AgentBase) { b.name = name })
}

// WithDescription sets the agent description.
func WithDescription(desc string) Option {
	return optionFunc(func(b *AgentBase) { b.description = desc })
}

// AgentBase is the base agent implementation that offers
// shared, default agent behavior. All agents should extend this
// base agent and configure the [AgentBase.Self] from the constructor
// [NewAgentBase].

// For example,
//
//		type MyCustomAgent struct {
//		   *AgentBase
//		   ....
//		}
//

Review comment (suggestion):
I'd suggest from the start for option to return error and check those in the constructor call. This way we'd get early validation at the constructor level already.

`type optionFunc func(*AgentBase) error`

Code after:
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package agent

import (
	"fmt"

	"github.com/google/adk-go"
)

// Option configures the Base agent.
type Option interface {
	apply2Base(*BaseAgent) error
}

type optionFunc func(*BaseAgent) error

func (o optionFunc) apply2Base(b *BaseAgent) error { return o(b) }

// WithName sets the agent name.
func WithName(name string) Option {
	return optionFunc(func(b *BaseAgent) error {
		b.name = name
		return nil
	})
}

// WithDescription sets the agent description.
func WithDescription(desc string) Option {
	return optionFunc(func(b *BaseAgent) error {
		b.description = desc
		return nil
	})
}

func WithSubAgents(agents ...adk.Agent) Option {
	return optionFunc(func(b *BaseAgent) error {
		return b.AddSubAgents(agents...)
	})
}

// BaseAgent is the base agent implementation that offers
Repository: google/adk-go
File: agent/base_agent.go (Go)

Code before:

func (a *BaseAgent) Name() string           { return a.name }
func (a *BaseAgent) Description() string    { return a.description }
func (a *BaseAgent) Parent() adk.Agent      { return a.parentAgent }
func (a *BaseAgent) SubAgents() []adk.Agent { return a.subAgents }
func (a *BaseAgent) Run(ctx context.Context, parentCtx *adk.InvocationContext) iter.Seq2[*adk.Event, error] {
	panic("unimplemented")
}

// TODO: Should we export it as Base() and include it in the interface?
// That will allows custom agents to wrap its BaseAgent instead of embedding.
func (a *BaseAgent) _base_() *BaseAgent { return a }

func (a *BaseAgent) findSubAgent(name string) (adk.Agent, bool) {
	for _, s := range a.subAgents {
		if s.Name() == name {
			return a, true
		}
	}
	return nil, false
}

// AddSubAgents adds the agents to the subagent list.
func (a *BaseAgent) AddSubAgents(agents ...adk.Agent) error {
	for _, subagent := range agents {
		// O(n^2) search, but n is small enough.
		if _, found := a.findSubAgent(subagent.Name()); found {
			return fmt.Errorf("cannot register multiple agents with the same name: %q", subagent.Name())
		}

		a.subAgents = append(a.subAgents, subagent)

		if s, ok := subagent.(interface{ _base_() *BaseAgent }); ok {
			if base := s._base_(); base != nil {
				if base.parentAgent != nil {
					return fmt.Errorf("Agent(%q) is already a subagent of Agent(%q)", base.Name(), base.parentAgent.Name())
				}
				base.parentAgent = a.self
			}
		}
	}
	return nil
}


Review comment (performance):
If you'd like you can unwrap it for O(n) check:

1. preprocess and get a map/set of a.subAgents (e.g. map[string]struct{})
2. check if a new agent is in this map already -> if yes return error

Code after:

// TODO: Should we export it as Base() and include it in the interface?
// That will allows custom agents to wrap its BaseAgent instead of embedding.
func (a *BaseAgent) _base_() *BaseAgent { return a }

func asBaseAgent(a adk.Agent) *BaseAgent {
	if b, ok := a.(interface{ _base_() *BaseAgent }); ok {
		return b._base_()
	}
	return nil
}

// AddSubAgents adds the agents to the subagent list.
func (a *BaseAgent) AddSubAgents(agents ...adk.Agent) error {
	names := map[string]bool{}
	for _, subagent := range a.subAgents {
		names[subagent.Name()] = true
	}
	// run sanity check (no duplicate name, no multiple parents)
	for _, subagent := range agents {
		name := subagent.Name()
		if names[name] {
			return fmt.Errorf("multiple subagents with the same name (%q) are not allowed", name)
		}
		if parent := subagent.Parent(); parent != nil {
			return fmt.Errorf("agent %q already has parent %q", name, parent.Name())
		}
		names[name] = true
	}

	// mutate.
	for _, subagent := range agents {
		a.subAgents = append(a.subAgents, subagent)
		if base := asBaseAgent(subagent); base != nil {
			base.parentAgent = a.self
		}
		// TODO: error if subagent is not a base agent?
	}
	return nil
}
Repository: google/adk-go
File: examples/userloop.go (Go)

Code before:
	"bufio"
	"context"
	"fmt"
	"log"
	"os"

	"google.golang.org/adk/agent"
	"google.golang.org/adk/runner"
	"google.golang.org/adk/sessionservice"
	"google.golang.org/genai"
)

func Run(ctx context.Context, rootAgent agent.Agent) {
	userID, appName := "test_user", "test_app"

	sessionService := sessionservice.Mem()

	session, err := sessionService.Create(ctx, &sessionservice.CreateRequest{
		AppName: appName,
		UserID:  userID,
	})
	if err != nil {
		panic(err)
	}

	agentRunner := runner.Runner{
		AppName:        appName,
		RootAgent:      rootAgent,
		SessionService: sessionService,
	}

	reader := bufio.NewReader(os.Stdin)

	for {
		fmt.Print("\nUser -> ")

		userInput, err := reader.ReadString('\n')
		if err != nil {
			log.Fatal(err)
		}

		userMsg := genai.NewContentFromText(userInput, genai.RoleUser)

		fmt.Print("\nAgent -> ")
		for event, err := range agentRunner.Run(ctx, userID, session.ID().SessionID, userMsg, &runner.AgentRunConfig{
			StreamingMode: runner.StreamingModeSSE,
		}) {
			if err != nil {
				fmt.Printf("\nAGENT_ERROR: %v\n", err)
			} else {
				for _, p := range event.LLMResponse.Content.Parts {

Review comment (suggestion):
Maybe call this `r`. The runner is the runtime, it's narrow to represent them as agent runners. Agent is just a component in the runtime like a tool or a model.

Code after:
	"bufio"
	"context"
	"fmt"
	"log"
	"os"

	"google.golang.org/adk/agent"
	"google.golang.org/adk/runner"
	"google.golang.org/adk/sessionservice"
	"google.golang.org/genai"
)

func Run(ctx context.Context, rootAgent agent.Agent) {
	userID, appName := "test_user", "test_app"

	sessionService := sessionservice.Mem()

	session, err := sessionService.Create(ctx, &sessionservice.CreateRequest{
		AppName: appName,
		UserID:  userID,
	})
	if err != nil {
		log.Fatalf("Failed to create the session service: %v", err)
	}

	r := runner.Runner{
		AppName:        appName,
		RootAgent:      rootAgent,
		SessionService: sessionService,
	}

	reader := bufio.NewReader(os.Stdin)

	for {
		fmt.Print("\nUser -> ")

		userInput, err := reader.ReadString('\n')
		if err != nil {
			log.Fatal(err)
		}

		userMsg := genai.NewContentFromText(userInput, genai.RoleUser)

		fmt.Print("\nAgent -> ")
		for event, err := range r.Run(ctx, userID, session.ID().SessionID, userMsg, &runner.RunConfig{
			StreamingMode: runner.StreamingModeSSE,
		}) {
			if err != nil {
				fmt.Printf("\nAGENT_ERROR: %v\n", err)
			} else {
				for _, p := range event.LLMResponse.Content.Parts {
Repository: google/adk-go
File: agent/workflowagents/parallelagent/agent.go (Go)

Code before:

		wg.Add(1)
		go func() {
			defer wg.Done()

			ctx := agent.NewContext(ctx, subAgent, ctx.UserContent(), ctx.Session(), branch)
			runSubAgent(ctx, subAgent, results)
		}()
	}

	go func() {
		wg.Wait()
		close(results)
	}()

	return func(yield func(*session.Event, error) bool) {
		for res := range results {
			if !yield(res.event, res.err) {
				break
			}
		}

		ctx.End()
	}
}

func runSubAgent(ctx agent.Context, agent agent.Agent, results chan<- result) {
	for event, err := range agent.Run(ctx) {
		select {
		case <-ctx.Done():
			return
		case results <- result{
			event: event,
			err:   err,
		}:
		}
	}
}

type result struct {
	event *session.Event
	err   error
}


Review comment (suggestion):
I feel like we need to drain the results channel if there are goroutines that are blocked on sending to `results`. 

Code after:
		if ctx.Branch() != "" {
			branch = fmt.Sprintf("%s.%s", ctx.Branch(), branch)
		}

		errGroup.Go(func() error {
			ctx := agent.NewContext(errGroupCtx, subAgent, ctx.UserContent(), ctx.Session(), branch)

			if err := runSubAgent(ctx, subAgent, resultsChan, doneChan); err != nil {
				return fmt.Errorf("failed to run sub-agent %q: %w", subAgent.Name(), err)
			}

			return nil
		})
	}

	go func() {
		_ = errGroup.Wait() // this error is already sent to the user via iterator
		close(resultsChan)
	}()

	return func(yield func(*session.Event, error) bool) {
		defer close(doneChan)

		for res := range resultsChan {
			if !yield(res.event, res.err) {
				break
			}
		}
	}
}

func runSubAgent(ctx agent.Context, agent agent.Agent, results chan<- result, done <-chan bool) error {
	for event, err := range agent.Run(ctx) {
		select {
		case <-done:
			return nil
		case <-ctx.Done():
			select {
			case <-done:
			case results <- result{
				err: ctx.Err(),
			}:
			}
			return ctx.Err()
		case results <- result{
			event: event,
			err:   err,
		}:
			if err != nil {
				return err
			}