ronantakizawa/codereview-bench huggingface dataset:
Repository: syncthing/syncthing
File: lib/protocol/protocol.go (Go)

Code before:
		// Use compression for large enough messages
		return msg.ProtoSize() >= compressionThreshold

	case CompressMetadata:
		_, isResponse := msg.(*Response)
		// Compress if it's large enough and not a response message
		return !isResponse && msg.ProtoSize() >= compressionThreshold

	default:
		panic("unknown compression setting")
	}
}

// Close is called when the connection is regularely closed and thus the Close
// BEP message is sent before terminating the actual connection. The error
// argument specifies the reason for closing the connection.
func (c *rawConnection) Close(err error) {
	c.once.Do(func() {
		done := make(chan struct{})
		c.sendClose <- asyncMessage{&Close{err.Error()}, done}
		<-done

		go c.commonClose(err)
	})
}

// close is called if there is an unexpected error during normal operation.
func (c *rawConnection) close(err error) {
	c.once.Do(func() {
		c.commonClose(err)
	})
}

// commonClose is a utility function that must only be called from within
// rawConnection.once.Do (i.e. in Close and close).
func (c *rawConnection) commonClose(err error) {
	l.Debugln("close due to", err)
	close(c.closed)

	c.awaitingMut.Lock()
	for i, ch := range c.awaiting {
		if ch != nil {
			close(ch)
			delete(c.awaiting, i)
		}
	}
	c.awaitingMut.Unlock()

	c.receiver.Closed(c, err)
}


Review comment (bug):
Some sort of comment on why we `go` this? I have a vague feeling it's because this might be called from a protocol callback and there are potential deadlock issues?

Code after:
		// Use compression for large enough messages
		return msg.ProtoSize() >= compressionThreshold

	case CompressMetadata:
		_, isResponse := msg.(*Response)
		// Compress if it's large enough and not a response message
		return !isResponse && msg.ProtoSize() >= compressionThreshold

	default:
		panic("unknown compression setting")
	}
}

// Close is called when the connection is regularely closed and thus the Close
// BEP message is sent before terminating the actual connection. The error
// argument specifies the reason for closing the connection.
func (c *rawConnection) Close(err error) {
	c.once.Do(func() {
		done := make(chan struct{})
		c.sendClose <- asyncMessage{&Close{err.Error()}, done}
		<-done

		// No more sends are necessary, therefore closing the underlying
		// connection can happen at the same time as the internal cleanup.
		// And this prevents a potential deadlock due to calling c.receiver.Closed
		go c.commonClose(err)
	})
}

// internalClose is called if there is an unexpected error during normal operation.
func (c *rawConnection) internalClose(err error) {
	c.once.Do(func() {
		c.commonClose(err)
	})
}

// commonClose is a utility function that must only be called from within
// rawConnection.once.Do (i.e. in Close and close).
func (c *rawConnection) commonClose(err error) {
	l.Debugln("close due to", err)
	close(c.closed)

	c.awaitingMut.Lock()
	for i, ch := range c.awaiting {
		if ch != nil {
			close(ch)
			delete(c.awaiting, i)
		}
	}
	c.awaitingMut.Unlock()
Repository: syncthing/syncthing
File: lib/db/instance.go (Go)

Code before:
			name := []byte(f.Name)
			gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
			t.removeFromGlobal(gk, folder, device, name, nil)
			t.Delete(dbi.Key())
			t.checkFlush()
			continue
		}

		if !fn(device, f) {
			return
		}
	}
}

func (db *instance) getFileDirty(folder, device, file []byte) (protocol.FileInfo, bool) {
	return db.getFileDirtyByKey(db.keyer.GenerateDeviceFileKey(nil, folder, device, file))
}

func (db *instance) getFileDirtyByKey(key []byte) (protocol.FileInfo, bool) {
	if f, ok := db.getFileDirtyTrunc(key, false); ok {
		return f.(protocol.FileInfo), true
	}
	return protocol.FileInfo{}, false
}

func (db *instance) getFileDirtyTrunc(key []byte, trunc bool) (FileIntf, bool) {
	bs, err := db.Get(key, nil)
	if err == leveldb.ErrNotFound {
		return nil, false
	}
	if err != nil {
		l.Debugln("surprise error:", err)
		return nil, false
	}

	f, err := unmarshalTrunc(bs, trunc)
	if err != nil {
		l.Debugln("unmarshal error:", err)
		return nil, false
	}
	return f, true
}

func (db *instance) getGlobal(folder, file []byte, truncate bool) (FileIntf, bool) {
	t := db.newReadOnlyTransaction()
	defer t.close()

	_, _, f, ok := db.getGlobalInto(t, nil, nil, folder, file, truncate)
	return f, ok
}


Review comment (suggestion):
There are now only calls to `db.getFileDirty(...` left, so this function and `getFileDirtyByKey` could be removed.

Code after:
			name := []byte(f.Name)
			gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
			t.removeFromGlobal(gk, folder, device, name, nil)
			t.Delete(dbi.Key())
			t.checkFlush()
			continue
		}

		if !fn(device, f) {
			return
		}
	}
}

func (db *instance) getFileDirty(folder, device, file []byte) (protocol.FileInfo, bool) {
	key := db.keyer.GenerateDeviceFileKey(nil, folder, device, file)
	bs, err := db.Get(key, nil)
	if err == leveldb.ErrNotFound {
		return protocol.FileInfo{}, false
	}
	if err != nil {
		l.Debugln("surprise error:", err)
		return protocol.FileInfo{}, false
	}

	var f protocol.FileInfo
	if err := f.Unmarshal(bs); err != nil {
		l.Debugln("unmarshal error:", err)
		return protocol.FileInfo{}, false
	}
	return f, true
}

func (db *instance) getGlobal(folder, file []byte, truncate bool) (FileIntf, bool) {
	t := db.newReadOnlyTransaction()
	defer t.close()

	_, _, f, ok := db.getGlobalInto(t, nil, nil, folder, file, truncate)
	return f, ok
}

func (db *instance) getGlobalInto(t readOnlyTransaction, gk, dk, folder, file []byte, truncate bool) ([]byte, []byte, FileIntf, bool) {
	gk = db.keyer.GenerateGlobalVersionKey(gk, folder, file)

	bs, err := t.Get(gk, nil)
	if err != nil {
		return gk, dk, nil, false
	}

	vl, ok := unmarshalVersionList(bs)
	if !ok {
Repository: syncthing/syncthing
File: lib/db/instance.go (Go)

Code before:
func newInstance(ll *Lowlevel) *instance {
	return &instance{
		Lowlevel: ll,
		keyer:    newDefaultKeyer(ll.folderIdx, ll.deviceIdx),
	}
}

func (db *instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, meta *metadataTracker) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var fk []byte
	var gk []byte
	for _, f := range fs {
		name := []byte(f.Name)
		fk = db.keyer.GenerateDeviceFileKey(fk, folder, device, name)

		// Get and unmarshal the file entry. If it doesn't exist or can't be
		// unmarshalled we'll add it as a new entry.
		bs, err := t.Get(fk, nil)
		var ef FileInfoTruncated
		if err == nil {
			err = ef.Unmarshal(bs)
		}

		if unchanged(f, ef, err == nil) {
			continue
		}

		devID := protocol.DeviceIDFromBytes(device)
		if err == nil {
			meta.removeFile(devID, ef)
		}
		meta.addFile(devID, f)

		t.insertFile(fk, folder, device, f)

		gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
		t.updateGlobal(gk, folder, device, f, meta)

		// Write out and reuse the batch every few records, to avoid the batch
		// growing too large and thus allocating unnecessarily much memory.
		t.checkFlush()
	}
}

func (db *instance) addSequences(folder []byte, fs []protocol.FileInfo) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var sk []byte

Review comment (suggestion):
Passing the bool is weird. `err == nil && dbEquivalent(f, ef)` scans better imho

Code after:
func newInstance(ll *Lowlevel) *instance {
	return &instance{
		Lowlevel: ll,
		keyer:    newDefaultKeyer(ll.folderIdx, ll.deviceIdx),
	}
}

func (db *instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, meta *metadataTracker) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var fk []byte
	var gk []byte
	for _, f := range fs {
		name := []byte(f.Name)
		fk = db.keyer.GenerateDeviceFileKey(fk, folder, device, name)

		// Get and unmarshal the file entry. If it doesn't exist or can't be
		// unmarshalled we'll add it as a new entry.
		bs, err := t.Get(fk, nil)
		var ef FileInfoTruncated
		if err == nil {
			err = ef.Unmarshal(bs)
		}

		if err == nil && unchanged(f, ef) {
			continue
		}

		devID := protocol.DeviceIDFromBytes(device)
		if err == nil {
			meta.removeFile(devID, ef)
		}
		meta.addFile(devID, f)

		t.insertFile(fk, folder, device, f)

		gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
		t.updateGlobal(gk, folder, device, f, meta)

		// Write out and reuse the batch every few records, to avoid the batch
		// growing too large and thus allocating unnecessarily much memory.
		t.checkFlush()
	}
}

func (db *instance) addSequences(folder []byte, fs []protocol.FileInfo) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var sk []byte
Repository: syncthing/syncthing
File: lib/db/instance.go (Go)

Code before:

	"github.com/syncthing/syncthing/lib/protocol"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/iterator"
	"github.com/syndtr/goleveldb/leveldb/util"
)

type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator)

type instance struct {
	*Lowlevel
	keyer keyer
}

func newInstance(ll *Lowlevel) *instance {
	return &instance{
		Lowlevel: ll,
		keyer:    newDefaultKeyer(ll.folderIdx, ll.deviceIdx),
	}
}

func (db *instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, meta *metadataTracker) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var dk, gk, keyBuf []byte
	devID := protocol.DeviceIDFromBytes(device)
	for _, f := range fs {
		name := []byte(f.Name)
		dk = db.keyer.GenerateDeviceFileKey(dk, folder, device, name)

		ef, ok := t.getFileTrunc(dk, true)
		if ok && unchanged(f, ef) {
			continue
		}

		if ok {
			meta.removeFile(devID, ef)
		}
		meta.addFile(devID, f)

		t.insertFile(dk, folder, device, f)

		gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
		t.updateGlobal(gk, keyBuf, folder, device, f, meta)

		// Write out and reuse the batch every few records, to avoid the batch
		// growing too large and thus allocating unnecessarily much memory.
		t.checkFlush()
	}
}

Review comment (suggestion):
The same thing could be achieved with just two buffers and the naming isn't as succinct as possible (keyBuf vs just buf), but this choice will become more clear in the followup PR (with blockBuf). Having three buffers also is a bit safer, as then `dk` really only is used as device-file-key, not for other purposes too.

Code after:

	"github.com/syncthing/syncthing/lib/protocol"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/iterator"
	"github.com/syndtr/goleveldb/leveldb/util"
)

type deletionHandler func(t readWriteTransaction, folder, device, name []byte, dbi iterator.Iterator)

type instance struct {
	*Lowlevel
	keyer keyer
}

func newInstance(ll *Lowlevel) *instance {
	return &instance{
		Lowlevel: ll,
		keyer:    newDefaultKeyer(ll.folderIdx, ll.deviceIdx),
	}
}

func (db *instance) updateFiles(folder, device []byte, fs []protocol.FileInfo, meta *metadataTracker) {
	t := db.newReadWriteTransaction()
	defer t.close()

	var dk, gk, keyBuf []byte
	devID := protocol.DeviceIDFromBytes(device)
	for _, f := range fs {
		name := []byte(f.Name)
		dk = db.keyer.GenerateDeviceFileKey(dk, folder, device, name)

		ef, ok := t.getFileTrunc(dk, true)
		if ok && unchanged(f, ef) {
			continue
		}

		if ok {
			meta.removeFile(devID, ef)
		}
		meta.addFile(devID, f)

		t.insertFile(dk, folder, device, f)

		gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
		keyBuf, _ = t.updateGlobal(gk, keyBuf, folder, device, f, meta)

		// Write out and reuse the batch every few records, to avoid the batch
		// growing too large and thus allocating unnecessarily much memory.
		t.checkFlush()
	}
}
Repository: syncthing/syncthing
File: lib/db/namespaced.go (Go)

Code before:

package db

import (
	"encoding/binary"
	"time"

	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/util"
)

// NamespacedKV is a simple key-value store using a specific namespace within
// a leveldb.
type NamespacedKV struct {
	db        *Lowlevel
	prefix    []byte
	prefixLen int
}

// NewNamespacedKV returns a new NamespacedKV that lives in the namespace
// specified by the prefix.
func NewNamespacedKV(db *Lowlevel, prefix string) *NamespacedKV {
	return &NamespacedKV{
		db:        db,
		prefix:    []byte(prefix),
		prefixLen: len(prefix),
	}
}

// Reset removes all entries in this namespace.
func (n *NamespacedKV) Reset() {
	it := n.db.NewIterator(util.BytesPrefix(n.prefix), nil)
	defer it.Release()
	batch := new(leveldb.Batch)
	for it.Next() {
		batch.Delete(it.Key())
		if batch.Len() > batchFlushSize {
			if err := n.db.Write(batch, nil); err != nil {
				panic(err)
			}
			batch.Reset()
		}
	}
	if batch.Len() > 0 {
		if err := n.db.Write(batch, nil); err != nil {
			panic(err)
		}
	}
}

// PutInt64 stores a new int64. Any existing value (even if of another type)

Review comment (suggestion):
If this is just an optimization I think we don't need it. `len(prefix)` is already just a read of the length word in the slice header, even though it looks like a function call.

Code after:

package db

import (
	"encoding/binary"
	"time"

	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/util"
)

// NamespacedKV is a simple key-value store using a specific namespace within
// a leveldb.
type NamespacedKV struct {
	db     *Lowlevel
	prefix []byte
}

// NewNamespacedKV returns a new NamespacedKV that lives in the namespace
// specified by the prefix.
func NewNamespacedKV(db *Lowlevel, prefix string) *NamespacedKV {
	prefixBs := []byte(prefix)
	// After the conversion from string the cap will be larger than the len (in Go 1.11.5,
	// 32 bytes cap for small strings). We need to cut it down to ensure append() calls
	// on the prefix make a new allocation.
	prefixBs = prefixBs[:len(prefixBs):len(prefixBs)]
	return &NamespacedKV{
		db:     db,
		prefix: prefixBs,
	}
}

// Reset removes all entries in this namespace.
func (n *NamespacedKV) Reset() {
	it := n.db.NewIterator(util.BytesPrefix(n.prefix), nil)
	defer it.Release()
	batch := new(leveldb.Batch)
	for it.Next() {
		batch.Delete(it.Key())
		if batch.Len() > batchFlushSize {
			if err := n.db.Write(batch, nil); err != nil {
				panic(err)
			}
			batch.Reset()
		}
	}
	if batch.Len() > 0 {
		if err := n.db.Write(batch, nil); err != nil {
			panic(err)
		}
	}
Repository: syncthing/syncthing
File: lib/model/folder.go (Go)

Code before:

	if f.FSWatcherEnabled {
		if err != nil {
			f.stopWatch()
		} else {
			f.scheduleWatchRestart()
		}
	}

	f.stateTracker.setError(err)
}

func (f *folder) basePause() time.Duration {
	if f.PullerPauseS == 0 {
		return defaultPullerPause
	}
	return time.Duration(f.PullerPauseS) * time.Second
}

// resetPullFailTimer reschedules pull retrying with increasingly longer pauses
// between tries (with lower limit based on pull duration and fix upper limit).
func (f *folder) resetPullFailTimer(timer *time.Timer, pause, pullDur time.Duration) time.Duration {
	max := 60 * f.basePause()
	// Pause at least as long as the time the previous (failed) pull took
	for pullDur > pause && pause < max {
		pause *= 2
	}
	l.Infof("Folder %v isn't making sync progress - retrying in %v.", f.Description(), pause)
	timer.Reset(pause)
	if pause > max {
		return pause
	}
	return 2 * pause
}

func (f *folder) String() string {
	return fmt.Sprintf("%s/%s@%p", f.Type, f.folderID, f)
}

func (f *folder) newScanError(path string, err error) {
	f.scanErrorsMut.Lock()
	f.scanErrors = append(f.scanErrors, FileError{
		Err:  err.Error(),
		Path: path,
	})
	f.scanErrorsMut.Unlock()
}

func (f *folder) clearScanErrors(subDirs []string) {
	f.scanErrorsMut.Lock()
	defer f.scanErrorsMut.Unlock()

Review comment (suggestion):
This method is confusing. There are multiple opportunities to get a pause > max and in some cases we multiply the pause by four in one pass... 

I think it would be simpler to keep the easy sequence on `pause` (1, 2, 4 etc minutes up to max) and always sleep `pullDur + pause`? In this case we'll exceed max by at most `pullDur`, which is either negligible (normal case) or desirable (extreme case).

Code after:
	} else {
		l.Infoln("Cleared error on folder", f.Description())
	}

	if f.FSWatcherEnabled {
		if err != nil {
			f.stopWatch()
		} else {
			f.scheduleWatchRestart()
		}
	}

	f.stateTracker.setError(err)
}

func (f *folder) basePause() time.Duration {
	if f.PullerPauseS == 0 {
		return defaultPullerPause
	}
	return time.Duration(f.PullerPauseS) * time.Second
}

func (f *folder) String() string {
	return fmt.Sprintf("%s/%s@%p", f.Type, f.folderID, f)
}

func (f *folder) newScanError(path string, err error) {
	f.scanErrorsMut.Lock()
	f.scanErrors = append(f.scanErrors, FileError{
		Err:  err.Error(),
		Path: path,
	})
	f.scanErrorsMut.Unlock()
}

func (f *folder) clearScanErrors(subDirs []string) {
	f.scanErrorsMut.Lock()
	defer f.scanErrorsMut.Unlock()
	if len(subDirs) == 0 {
		f.scanErrors = nil
		return
	}
	filtered := f.scanErrors[:0]
outer:
	for _, fe := range f.scanErrors {
		for _, sub := range subDirs {
			if fe.Path == sub || fs.IsParent(fe.Path, sub) {
				continue outer
			}
		}
		filtered = append(filtered, fe)
Repository: syncthing/syncthing
File: lib/model/model.go (Go)

Code before:
	case device.MaxRequestKiB == 0:
		m.connRequestLimiters[deviceID] = newByteSemaphore(1024 * defaultPullerPendingKiB)
	}

	m.helloMessages[deviceID] = hello

	event := map[string]string{
		"id":            deviceID.String(),
		"deviceName":    hello.DeviceName,
		"clientName":    hello.ClientName,
		"clientVersion": hello.ClientVersion,
		"type":          conn.Type(),
	}

	addr := conn.RemoteAddr()
	if addr != nil {
		event["addr"] = addr.String()
	}

	events.Default.Log(events.DeviceConnected, event)

	l.Infof(`Device %s client is "%s %s" named "%s" at %s`, deviceID, hello.ClientName, hello.ClientVersion, hello.DeviceName, conn)

	conn.Start()

	conn.ClusterConfig(cm)

	if (device.Name == "" || m.cfg.Options().OverwriteRemoteDevNames) && hello.DeviceName != "" {
		device.Name = hello.DeviceName
		m.cfg.SetDevice(device)
		m.cfg.Save()
	}

	m.deviceWasSeen(deviceID)
}

func (m *model) DownloadProgress(device protocol.DeviceID, folder string, updates []protocol.FileDownloadProgressUpdate) {
	m.fmut.RLock()
	cfg, ok := m.folderCfgs[folder]
	m.fmut.RUnlock()

	if !ok || cfg.DisableTempIndexes || !cfg.SharedWith(device) {
		return
	}

	m.pmut.RLock()
	m.deviceDownloads[device].Update(folder, updates)
	state := m.deviceDownloads[device].GetBlockCounts(folder)
	m.pmut.RUnlock()

	events.Default.Log(events.RemoteDownloadProgress, map[string]interface{}{

Review comment (suggestion):
Only thing that makes me a little bit uneasy about this is that ClusterConfig() is a potentially blocking call... Although, at this point, the writer loop should be up since Start() and be reading the outbox channel that ClusterConfig() needs to send on, so hopefully it's fine.

Code after:

	event := map[string]string{
		"id":            deviceID.String(),
		"deviceName":    hello.DeviceName,
		"clientName":    hello.ClientName,
		"clientVersion": hello.ClientVersion,
		"type":          conn.Type(),
	}

	addr := conn.RemoteAddr()
	if addr != nil {
		event["addr"] = addr.String()
	}

	events.Default.Log(events.DeviceConnected, event)

	l.Infof(`Device %s client is "%s %s" named "%s" at %s`, deviceID, hello.ClientName, hello.ClientVersion, hello.DeviceName, conn)

	conn.Start()
	m.pmut.Unlock()

	// Acquires fmut, so has to be done outside of pmut.
	cm := m.generateClusterConfig(deviceID)
	conn.ClusterConfig(cm)

	if (device.Name == "" || m.cfg.Options().OverwriteRemoteDevNames) && hello.DeviceName != "" {
		device.Name = hello.DeviceName
		m.cfg.SetDevice(device)
		m.cfg.Save()
	}

	m.deviceWasSeen(deviceID)
}

func (m *model) DownloadProgress(device protocol.DeviceID, folder string, updates []protocol.FileDownloadProgressUpdate) {
	m.fmut.RLock()
	cfg, ok := m.folderCfgs[folder]
	m.fmut.RUnlock()

	if !ok || cfg.DisableTempIndexes || !cfg.SharedWith(device) {
		return
	}

	m.pmut.RLock()
	m.deviceDownloads[device].Update(folder, updates)
	state := m.deviceDownloads[device].GetBlockCounts(folder)
	m.pmut.RUnlock()

	events.Default.Log(events.RemoteDownloadProgress, map[string]interface{}{
		"device": device.String(),
		"folder": folder,
Repository: syncthing/syncthing
File: lib/protocol/protocol.go (Go)

Code before:
	}
	close(c.sentClusterConfig)
}

func (c *rawConnection) Closed() bool {
	select {
	case <-c.closed:
		return true
	default:
		return false
	}
}

// DownloadProgress sends the progress updates for the files that are currently being downloaded.
func (c *rawConnection) DownloadProgress(folder string, updates []FileDownloadProgressUpdate) {
	c.send(&DownloadProgress{
		Folder:  folder,
		Updates: updates,
	}, nil)
}

func (c *rawConnection) ping() bool {
	return c.send(&Ping{}, nil)
}

type errorMessage struct {
	msg message
	err error
}

func (c *rawConnection) readerLoop() error {
	fourByteBuf := make([]byte, 4)
	inbox := make(chan errorMessage)

	// Reading from the wire may block until the underlying connection is closed.
	go func() {
		for {
			msg, err := c.readMessage(fourByteBuf)
			select {
			case inbox <- errorMessage{msg: msg, err: err}:
			case <-c.closed:
				return
			}
		}
	}()

	state := stateInitial
	var errMsg errorMessage
	for {
		select {
		case errMsg = <-inbox:

Review comment (suggestion):
readResponse? This feels like an actual error message not the conjunction of two types independently.

Code after:
	}
	close(c.sentClusterConfig)
}

func (c *rawConnection) Closed() bool {
	select {
	case <-c.closed:
		return true
	default:
		return false
	}
}

// DownloadProgress sends the progress updates for the files that are currently being downloaded.
func (c *rawConnection) DownloadProgress(folder string, updates []FileDownloadProgressUpdate) {
	c.send(&DownloadProgress{
		Folder:  folder,
		Updates: updates,
	}, nil)
}

func (c *rawConnection) ping() bool {
	return c.send(&Ping{}, nil)
}

type messageWithError struct {
	msg message
	err error
}

func (c *rawConnection) readerLoop() error {
	fourByteBuf := make([]byte, 4)
	inbox := make(chan messageWithError)

	// Reading from the wire may block until the underlying connection is closed.
	go func() {
		for {
			msg, err := c.readMessage(fourByteBuf)
			select {
			case inbox <- messageWithError{msg: msg, err: err}:
			case <-c.closed:
				return
			}
		}
	}()

	state := stateInitial
	var msgWithErr messageWithError
	for {
		select {
		case msgWithErr = <-inbox: