<commit_before>package dev

import (
	"io"
	"sync"
	"time"

	"golang.org/x/net/context"
)

// RegisterTestServer is a basic register server that in addition also can
// signal when a read or write has completed.
type RegisterTestServer interface {
	RegisterServer
	ReadExecuted()
	WriteExecuted()
}

// RegisterServerBasic represents a single State register.
type RegisterServerBasic struct {
	sync.RWMutex
	state State

	readExecutedChan  chan struct{}
	writeExecutedChan chan struct{}
}

// NewRegisterBasic returns a new basic register server.
func NewRegisterBasic() *RegisterServerBasic {
	return &RegisterServerBasic{
		// Use an appropriate larger buffer size if we construct test
		// scenarios where it's needed.
		writeExecutedChan: make(chan struct{}, 32),
		readExecutedChan:  make(chan struct{}, 32),
	}
}

// NewRegisterBasicWithState returns a new basic register server with an initial
// state set.
func NewRegisterBasicWithState(state *State) *RegisterServerBasic {
	return &RegisterServerBasic{
		state: *state,
		// Use an appropriate larger buffer size if we construct test
		// scenarios where it's needed.
		writeExecutedChan: make(chan struct{}, 32),
		readExecutedChan:  make(chan struct{}, 32),
	}
}

func (r *RegisterServerBasic) Read(ctx context.Context, rq *ReadRequest) (*State, error) {
	r.RLock()
	defer r.RUnlock()
	r.readExecutedChan <- struct{}{}
	return &State{Value: r.state.Value, Timestamp: r.state.Timestamp}, nil
}

func (r *RegisterServerBasic) Write(ctx context.Context, s *State) (*WriteResponse, error) {
	r.Lock()
	defer r.Unlock()
	writeResp := &WriteResponse{}
	if s.Timestamp > r.state.Timestamp {
		r.state = *s
		writeResp.New = true
	}
	r.writeExecutedChan <- struct{}{}
	return writeResp, nil
}

// WriteAsync implements the WriteAsync method from the RegisterServer interface.
func (r *RegisterServerBasic) WriteAsync(stream Register_WriteAsyncServer) error {
	for {
		state, err := stream.Recv()
		if err == io.EOF {
			return stream.SendAndClose(&Empty{})
		}
		if err != nil {
			return err
		}
		_, err = r.Write(context.Background(), state)
		if err != nil {
			return err
		}
	}
}

// ReadNoQC implements the ReadNoQC method from the RegisterServer interface.
func (r *RegisterServerBasic) ReadNoQC(ctx context.Context, rq *ReadRequest) (*State, error) {
	return r.Read(ctx, rq)
}

// ReadTwo implements the ReadTwo method from the RegisterServer interface.
func (r *RegisterServerBasic) ReadTwo(rq *ReadRequest, rrts Register_ReadTwoServer) error {
	err := rrts.Send(&r.state)
	if err != nil {
		return err
	}
	return nil
}

// ReadExecuted returns when r has has completed a read.
func (r *RegisterServerBasic) ReadExecuted() {
	<-r.readExecutedChan<commit_msg>dev: simplify a server handler<commit_after>package dev

import (
	"io"
	"sync"
	"time"

	"golang.org/x/net/context"
)

// RegisterTestServer is a basic register server that in addition also can
// signal when a read or write has completed.
type RegisterTestServer interface {
	RegisterServer
	ReadExecuted()
	WriteExecuted()
}

// RegisterServerBasic represents a single State register.
type RegisterServerBasic struct {
	sync.RWMutex
	state State

	readExecutedChan  chan struct{}
	writeExecutedChan chan struct{}
}

// NewRegisterBasic returns a new basic register server.
func NewRegisterBasic() *RegisterServerBasic {
	return &RegisterServerBasic{
		// Use an appropriate larger buffer size if we construct test
		// scenarios where it's needed.
		writeExecutedChan: make(chan struct{}, 32),
		readExecutedChan:  make(chan struct{}, 32),
	}
}

// NewRegisterBasicWithState returns a new basic register server with an initial
// state set.
func NewRegisterBasicWithState(state *State) *RegisterServerBasic {
	return &RegisterServerBasic{
		state: *state,
		// Use an appropriate larger buffer size if we construct test
		// scenarios where it's needed.
		writeExecutedChan: make(chan struct{}, 32),
		readExecutedChan:  make(chan struct{}, 32),
	}
}

func (r *RegisterServerBasic) Read(ctx context.Context, rq *ReadRequest) (*State, error) {
	r.RLock()
	defer r.RUnlock()
	r.readExecutedChan <- struct{}{}
	return &State{Value: r.state.Value, Timestamp: r.state.Timestamp}, nil
}

func (r *RegisterServerBasic) Write(ctx context.Context, s *State) (*WriteResponse, error) {
	r.Lock()
	defer r.Unlock()
	writeResp := &WriteResponse{}
	if s.Timestamp > r.state.Timestamp {
		r.state = *s
		writeResp.New = true
	}
	r.writeExecutedChan <- struct{}{}
	return writeResp, nil
}

// WriteAsync implements the WriteAsync method from the RegisterServer interface.
func (r *RegisterServerBasic) WriteAsync(stream Register_WriteAsyncServer) error {
	for {
		state, err := stream.Recv()
		if err == io.EOF {
			return stream.SendAndClose(&Empty{})
		}
		if err != nil {
			return err
		}
		_, err = r.Write(context.Background(), state)
		if err != nil {
			return err
		}
	}
}

// ReadNoQC implements the ReadNoQC method from the RegisterServer interface.
func (r *RegisterServerBasic) ReadNoQC(ctx context.Context, rq *ReadRequest) (*State, error) {
	return r.Read(ctx, rq)
}

// ReadTwo implements the ReadTwo method from the RegisterServer interface.
func (r *RegisterServerBasic) ReadTwo(rq *ReadRequest, rrts Register_ReadTwoServer) error {
	return rrts.Send(&r.state)
}

// ReadExecuted returns when r has has completed a read.
func (r *RegisterServerBasic) ReadExecuted() {
	<-r.readExecutedChan
{'fraction_non_alphanumeric': 0.10616691285081241, 'fraction_numerical': 0.0029542097488921715, 'mean_word_length': 4.433299899699097, 'pattern_counts': {'":': 0, '<': 9, '<?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': '6163149', 'n_tokens_mistral': 1767, 'n_tokens_neox': 1650, 'n_words': 536}