0f5a2f4ef5
* enable more linters * Run gofmt -s * run goimports * run unconvert * fix naked return * fix misspellings * fix ineffectual assigments * fix missing declaration * don't use deprecated grpc.Errof * check errors in tests * run gofmt -w -r "assert.Nil(err) -> assert.NoError(err)" * fix directory permissions * don't use nil Context * simplify boolean expressions * use bytes.Equal instead of bytes.Compare * merge variable declarations, remove redundant returns * fix some golint errors * run goimports * handle more errors * delete empty TestMain * delete empty TestMain * ignore examples for now * fix lint errors * remove unused values * more fixes * run gofmt -w -s . * add more comments * fix naming * more lint fixes * try switching travis to go1.11 * fix unnecessary conversions * fix deprecated methods * use go1.10 and disable gofmt/goimports for now * switch to 1.10 * don't re-enable gofmt and goimports * switch covermode to atomic because of -race * gofmt
99 lines
2.2 KiB
Go
99 lines
2.2 KiB
Go
// Copyright (C) 2018 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log"
|
|
|
|
"github.com/zeebo/errs"
|
|
"storj.io/storj/pkg/piecestore"
|
|
"storj.io/storj/pkg/utils"
|
|
pb "storj.io/storj/protos/piecestore"
|
|
)
|
|
|
|
// OK - Success!
|
|
const OK = "OK"
|
|
|
|
// StoreError is a type of error for failures in Server.Store()
|
|
var StoreError = errs.Class("store error")
|
|
|
|
// Store incoming data using piecestore
|
|
func (s *Server) Store(reqStream pb.PieceStoreRoutes_StoreServer) (err error) {
|
|
ctx := reqStream.Context()
|
|
defer mon.Task()(&ctx)(&err)
|
|
// Receive id/ttl
|
|
recv, err := reqStream.Recv()
|
|
if err != nil {
|
|
return StoreError.Wrap(err)
|
|
}
|
|
if recv == nil {
|
|
return StoreError.New("Error receiving Piece metadata")
|
|
}
|
|
|
|
pd := recv.GetPiecedata()
|
|
if pd == nil {
|
|
return StoreError.New("PieceStore message is nil")
|
|
}
|
|
|
|
log.Printf("Storing %s...", pd.GetId())
|
|
|
|
if pd.GetId() == "" {
|
|
return StoreError.New("Piece ID not specified")
|
|
}
|
|
|
|
// If we put in the database first then that checks if the data already exists
|
|
if err = s.DB.AddTTLToDB(pd.GetId(), pd.GetExpirationUnixSec()); err != nil {
|
|
return StoreError.New("Failed to write expiration data to database")
|
|
}
|
|
|
|
total, err := s.storeData(ctx, reqStream, pd.GetId())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Printf("Successfully stored %s.", pd.GetId())
|
|
|
|
return reqStream.SendAndClose(&pb.PieceStoreSummary{Message: OK, TotalReceived: total})
|
|
}
|
|
|
|
func (s *Server) storeData(ctx context.Context, stream pb.PieceStoreRoutes_StoreServer, id string) (total int64, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
// Delete data if we error
|
|
defer func() {
|
|
if err != nil && err != io.EOF {
|
|
if err = s.deleteByID(id); err != nil {
|
|
log.Printf("Failed on deleteByID in Store: %s", err.Error())
|
|
}
|
|
}
|
|
}()
|
|
|
|
// Initialize file for storing data
|
|
storeFile, err := pstore.StoreWriter(id, s.DataDir)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
defer utils.LogClose(storeFile)
|
|
|
|
reader := NewStreamReader(s, stream)
|
|
|
|
defer func() {
|
|
err := s.DB.WriteBandwidthAllocToDB(reader.bandwidthAllocation)
|
|
if err != nil {
|
|
log.Printf("WriteBandwidthAllocToDB Error: %s\n", err.Error())
|
|
}
|
|
}()
|
|
|
|
total, err = io.Copy(storeFile, reader)
|
|
|
|
if err != nil && err != io.EOF {
|
|
return 0, err
|
|
}
|
|
|
|
return total, nil
|
|
}
|