LogDog: BigTable batching schema.

server/logdog/storage/memory/memory.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package memory
 
 import (
         "errors"
         "sync"
         "time"
@@ skipping 42 matching lines @@
 
 // Config implements storage.Storage.
 func (s *Storage) Config(cfg storage.Config) error {
         return s.run(func() error {
                 s.MaxLogAge = cfg.MaxLogAge
                 return nil
         })
 }
 
 // Put implements storage.Storage.
-func (s *Storage) Put(req *storage.PutRequest) error {
+func (s *Storage) Put(req storage.PutRequest) error {
         return s.run(func() error {
                 ls := s.getLogStreamLocked(req.Path, true)
 
-                if _, ok := ls.logs[req.Index]; ok {
-                        return storage.ErrExists
-                }
+                for i, v := range req.Values {
+                        index := req.Index + types.MessageIndex(i)
+                        if _, ok := ls.logs[index]; ok {
+                                return storage.ErrExists
+                        }
 
-                ls.logs[req.Index] = []byte(req.Value)
-                if req.Index > ls.latestIndex {
-                        ls.latestIndex = req.Index
+                        clone := make([]byte, len(v))
+                        copy(clone, v)
+                        ls.logs[index] = clone
+                        if index > ls.latestIndex {
+                                ls.latestIndex = index
+                        }
                 }
                 return nil
         })
 }
 
 // Get implements storage.Storage.
-func (s *Storage) Get(req *storage.GetRequest, cb storage.GetCallback) error {
+func (s *Storage) Get(req storage.GetRequest, cb storage.GetCallback) error {
         recs := []*rec(nil)
         err := s.run(func() error {
                 ls := s.getLogStreamLocked(req.Path, false)
                 if ls == nil {
                         return storage.ErrDoesNotExist
                 }
 
                 limit := len(ls.logs)
                 if req.Limit > 0 && req.Limit < limit {
                         limit = req.Limit
@@ skipping 100 matching lines @@
                 }
 
                 if s.streams == nil {
                         s.streams = map[types.StreamPath]*logStream{}
                 }
                 s.streams[p] = ls
         }
 
         return ls
 }