Chunk Reader/Writer interface
diff --git a/chunk/chunk.go b/chunk/chunk.go
new file mode 100644
index 0000000..8d55fe0
--- /dev/null
+++ b/chunk/chunk.go
@@ -0,0 +1,35 @@
+package chunk
+
+import "bytes"
+import "io"
+
+type Chunk interface {
+ SizeBytes() int
+ ReadSeeker() io.ReadSeeker
+ Writer() io.Writer
+}
+
+type InMemoryChunk struct {
+ payload *[]byte
+}
+
+func NewEmptyInMemoryChunk(sizeBytes int) Chunk {
+ payload := make([]byte, sizeBytes)
+ return &InMemoryChunk{payload: &payload}
+}
+
+func NewInMemoryChunk(p *[]byte) Chunk {
+ return &InMemoryChunk{payload: p}
+}
+
+func (c *InMemoryChunk) SizeBytes() int {
+ return len(*c.payload)
+}
+
+func (c *InMemoryChunk) ReadSeeker() io.ReadSeeker {
+ return bytes.NewReader(*c.payload)
+}
+
+func (c *InMemoryChunk) Writer() io.Writer {
+ return bytes.NewBuffer(*c.payload)
+}
diff --git a/chunk/replicator.go b/chunk/replicator.go
new file mode 100644
index 0000000..f5543a0
--- /dev/null
+++ b/chunk/replicator.go
@@ -0,0 +1,10 @@
+package chunk
+
+import "io"
+
+func Replicate(from, to Chunk) (n int, err error) {
+ src := from.ReadSeeker()
+ src.Seek(int64(to.SizeBytes()), io.SeekStart)
+ m, err := io.Copy(to.Writer(), src)
+ return int(m), err
+}
diff --git a/chunk/server.go b/chunk/server.go
index c35d761..315ec00 100644
--- a/chunk/server.go
+++ b/chunk/server.go
@@ -1,6 +1,8 @@
package chunk
+import "bytes"
import "context"
+import "io"
import "sync"
import "pcloud/api"
@@ -15,7 +17,7 @@
func (s *ChunkServer) ListChunks(
ctx context.Context,
- request *api.ListChunksRequest) (*api.ListChunksResponse, error) {
+ req *api.ListChunksRequest) (*api.ListChunksResponse, error) {
resp := api.ListChunksResponse{}
s.chunks.Range(func(k, v interface{}) bool {
resp.ChunkId = append(resp.ChunkId, k.(string))
@@ -26,16 +28,34 @@
func (s *ChunkServer) ReadChunk(
ctx context.Context,
- request *api.ReadChunkRequest) (*api.ReadChunkResponse, error) {
- if data, ok := s.chunks.Load(request.ChunkId); ok {
- return &api.ReadChunkResponse{Data: data.([]byte)}, nil
+ req *api.ReadChunkRequest) (resp *api.ReadChunkResponse, err error) {
+ if value, ok := s.chunks.Load(req.ChunkId); ok {
+ chunk := value.(Chunk)
+ src := chunk.ReadSeeker()
+ if req.Offset != 0 {
+ _, err = src.Seek(int64(req.Offset), io.SeekStart)
+ if err != nil {
+ return
+ }
+ }
+ var dst bytes.Buffer
+ if req.NumBytes != 0 {
+ _, err = io.CopyN(&dst, src, int64(req.NumBytes))
+ } else {
+ _, err = io.Copy(&dst, src)
+ }
+ if err == nil {
+ resp = &api.ReadChunkResponse{Data: dst.Bytes()}
+ }
}
- return nil, nil
+ return
}
func (s *ChunkServer) StoreChunk(
ctx context.Context,
- request *api.StoreChunkRequest) (*api.StoreChunkResponse, error) {
- s.chunks.Store(request.ChunkId, request.Data)
+ req *api.StoreChunkRequest) (*api.StoreChunkResponse, error) {
+ data := req.Data
+ chunk := NewInMemoryChunk(&data)
+ s.chunks.Store(req.ChunkId, chunk)
return &api.StoreChunkResponse{}, nil
}
diff --git a/chunk/server_test.go b/chunk/server_test.go
new file mode 100644
index 0000000..834de89
--- /dev/null
+++ b/chunk/server_test.go
@@ -0,0 +1,75 @@
+package chunk
+
+import "context"
+import "bytes"
+import "testing"
+
+import "pcloud/api"
+
+func TestStoreChunk(t *testing.T) {
+ s := NewChunkServer()
+ _, err := s.StoreChunk(context.Background(), &api.StoreChunkRequest{
+ ChunkId: "foo",
+ Data: []byte("hello world")})
+ if err != nil {
+ t.Error(err)
+ }
+}
+
+func TestStoreAndReadChunk(t *testing.T) {
+ s := NewChunkServer()
+ _, err := s.StoreChunk(context.Background(), &api.StoreChunkRequest{
+ ChunkId: "foo",
+ Data: []byte("hello world")})
+ if err != nil {
+ t.Error(err)
+ }
+ resp, err := s.ReadChunk(context.Background(), &api.ReadChunkRequest{
+ ChunkId: "foo"})
+ if err != nil {
+ t.Error(err)
+ }
+ if bytes.Compare(resp.Data, []byte("hello world")) != 0 {
+ t.Errorf("Expected: %s\nGot: %s\n", "hello world", resp.Data)
+ }
+}
+
+func TestReadWithOffsets(t *testing.T) {
+ s := NewChunkServer()
+ _, err := s.StoreChunk(context.Background(), &api.StoreChunkRequest{
+ ChunkId: "foo",
+ Data: []byte("hello world")})
+ if err != nil {
+ t.Error(err)
+ }
+ resp, err := s.ReadChunk(context.Background(), &api.ReadChunkRequest{
+ ChunkId: "foo",
+ Offset: 0,
+ NumBytes: 2})
+ if err != nil {
+ t.Error(err)
+ }
+ if bytes.Compare(resp.Data, []byte("he")) != 0 {
+ t.Errorf("Expected: %s\nGot: %s\n", "he", resp.Data)
+ }
+ resp, err = s.ReadChunk(context.Background(), &api.ReadChunkRequest{
+ ChunkId: "foo",
+ Offset: 2,
+ NumBytes: 2})
+ if err != nil {
+ t.Error(err)
+ }
+ if bytes.Compare(resp.Data, []byte("ll")) != 0 {
+ t.Errorf("Expected: %s\nGot: %s\n", "ll", resp.Data)
+ }
+ resp, err = s.ReadChunk(context.Background(), &api.ReadChunkRequest{
+ ChunkId: "foo",
+ Offset: 4})
+ if err != nil {
+ t.Error(err)
+ }
+ if bytes.Compare(resp.Data, []byte("o world")) != 0 {
+ t.Errorf("Expected: %s\nGot: %s\n", "o world", resp.Data)
+ }
+
+}
diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml
index 983f892..1818038 100644
--- a/k8s/deployment.yaml
+++ b/k8s/deployment.yaml
@@ -40,7 +40,7 @@
volumes:
- name: code
hostPath:
- path: "/Users/lekva/dev/src/pcloud"
+ path: "/Users/lekva/dev/go/src/pcloud"
---
kind: Deployment
apiVersion: apps/v1
@@ -75,4 +75,4 @@
volumes:
- name: code
hostPath:
- path: "/Users/lekva/dev/src/pcloud"
+ path: "/Users/lekva/dev/go/src/pcloud"