feat(bsr): improve error handling
This commit is contained in:
1 parent
4488f5f8cd
commit
e29b844131
4 files changed
+57
-16
No files matched your search
+16
-4
@@ -246,7 +246,10 @@ func (c *Connection) NewMessagesWriter(ctx context.Context, dir Direction) (io.W
|
||||
}
|
||||
|
||||
messagesName := fmt.Sprintf(messagesFile, dir.String())
|
||||
c.container.WriteMeta(ctx, "messages", dir.String())
|
||||
_, err := c.container.WriteMeta(ctx, "messages", dir.String())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m, err := c.container.create(ctx, messagesName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -265,7 +268,10 @@ func (c *Connection) NewRequestsWriter(ctx context.Context, dir Direction) (io.W
|
||||
}
|
||||
|
||||
requestName := fmt.Sprintf(requestsFile, dir.String())
|
||||
c.container.WriteMeta(ctx, "requests", dir.String())
|
||||
_, err := c.container.WriteMeta(ctx, "requests", dir.String())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m, err := c.container.create(ctx, requestName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -308,7 +314,10 @@ func (c *Channel) NewMessagesWriter(ctx context.Context, dir Direction) (io.Writ
|
||||
}
|
||||
|
||||
messagesName := fmt.Sprintf(messagesFile, dir.String())
|
||||
c.container.WriteMeta(ctx, "messages", dir.String())
|
||||
_, err := c.container.WriteMeta(ctx, "messages", dir.String())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m, err := c.container.create(ctx, messagesName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -327,7 +336,10 @@ func (c *Channel) NewRequestsWriter(ctx context.Context, dir Direction) (io.Writ
|
||||
}
|
||||
|
||||
requestName := fmt.Sprintf(requestsFile, dir.String())
|
||||
c.container.WriteMeta(ctx, "requests", dir.String())
|
||||
_, err := c.container.WriteMeta(ctx, "requests", dir.String())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m, err := c.container.create(ctx, requestName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -72,7 +72,10 @@ func (e ChunkEncoder) Encode(ctx context.Context, c Chunk) (int, error) {
|
||||
if _, err := compressor.Write(data); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
compressor.Close()
|
||||
err = compressor.Close()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
length := buf.Len()
|
||||
|
||||
t := c.GetTimestamp().marshal()
|
||||
@@ -86,7 +89,10 @@ func (e ChunkEncoder) Encode(ctx context.Context, c Chunk) (int, error) {
|
||||
crced = append(crced, buf.Bytes()...)
|
||||
|
||||
crc := crc32.NewIEEE()
|
||||
crc.Write(crced)
|
||||
_, err = crc.Write(crced)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
d := make([]byte, 0, chunkBaseSize+length+crcSize)
|
||||
d = binary.BigEndian.AppendUint32(d, uint32(length))
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
// Package fstest provides test implementations of a the fs interfaces.
|
||||
// Package fstest provides test implementations of the fs interfaces.
|
||||
package fstest
|
||||
|
||||
import (
|
||||
@@ -20,7 +20,8 @@ import (
|
||||
type MemFS struct {
|
||||
Containers map[string]*MemContainer
|
||||
|
||||
newFunc NewFunc
|
||||
newFunc NewFunc
|
||||
readOnly bool
|
||||
}
|
||||
|
||||
// NewMemFS creates a MemFS. It supports WithNewFunc.
|
||||
@@ -30,6 +31,7 @@ func NewMemFS(options ...Option) *MemFS {
|
||||
return &MemFS{
|
||||
Containers: make(map[string]*MemContainer),
|
||||
newFunc: opts.withNewFunc,
|
||||
readOnly: opts.withReadOnly,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -45,9 +47,10 @@ func (m *MemFS) New(ctx context.Context, n string) (storage.Container, error) {
|
||||
return nil, fmt.Errorf("container %s already exists", n)
|
||||
}
|
||||
c := &MemContainer{
|
||||
Name: n,
|
||||
Sub: make(map[string]*MemContainer),
|
||||
Files: make(map[string]*MemFile),
|
||||
Name: n,
|
||||
Sub: make(map[string]*MemContainer),
|
||||
Files: make(map[string]*MemFile),
|
||||
readOnly: m.readOnly,
|
||||
}
|
||||
m.Containers[n] = c
|
||||
return c, nil
|
||||
@@ -60,7 +63,8 @@ type MemContainer struct {
|
||||
Sub map[string]*MemContainer
|
||||
Files map[string]*MemFile
|
||||
|
||||
Closed bool
|
||||
Closed bool
|
||||
readOnly bool
|
||||
|
||||
sync.RWMutex
|
||||
}
|
||||
@@ -82,9 +86,10 @@ func (m *MemContainer) Create(_ context.Context, n string) (storage.File, error)
|
||||
return nil, fmt.Errorf("file %s already exists", n)
|
||||
}
|
||||
f := &MemFile{
|
||||
name: n,
|
||||
Buf: bytes.NewBuffer([]byte{}),
|
||||
mode: sfs.ModeAppend,
|
||||
name: n,
|
||||
Buf: bytes.NewBuffer([]byte{}),
|
||||
mode: sfs.ModeAppend,
|
||||
ReadOnly: m.readOnly,
|
||||
}
|
||||
m.Files[n] = f
|
||||
return f, nil
|
||||
@@ -166,7 +171,9 @@ type MemFile struct {
|
||||
statFunc StatFunc
|
||||
closeFunc CloseFunc
|
||||
|
||||
Closed bool
|
||||
Closed bool
|
||||
ReadOnly bool
|
||||
OutOfSpace bool
|
||||
|
||||
sync.RWMutex
|
||||
}
|
||||
@@ -244,6 +251,13 @@ func (m *MemFile) Write(p []byte) (n int, err error) {
|
||||
return 0, fmt.Errorf("write on closed file")
|
||||
}
|
||||
|
||||
if m.ReadOnly {
|
||||
return 0, fmt.Errorf("write on read-only file")
|
||||
}
|
||||
|
||||
if m.OutOfSpace {
|
||||
return 0, fmt.Errorf("write failed, no space left on device")
|
||||
}
|
||||
defer func() {
|
||||
m.modtime = time.Now()
|
||||
}()
|
||||
|
||||
@@ -39,6 +39,7 @@ type options struct {
|
||||
withNewFunc NewFunc
|
||||
withCloseFunc CloseFunc
|
||||
withStatFunc StatFunc
|
||||
withReadOnly bool
|
||||
}
|
||||
|
||||
func getDefaultOptions() options {
|
||||
@@ -46,6 +47,7 @@ func getDefaultOptions() options {
|
||||
withNewFunc: nil,
|
||||
withCloseFunc: nil,
|
||||
withStatFunc: nil,
|
||||
withReadOnly: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -69,3 +71,10 @@ func WithStatFunc(f StatFunc) Option {
|
||||
o.withStatFunc = f
|
||||
}
|
||||
}
|
||||
|
||||
// WithReadOnly is used to set a read only bool
|
||||
func WithReadOnly(b bool) Option {
|
||||
return func(o *options) {
|
||||
o.withReadOnly = b
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user