143 lines
3.3 KiB
Go
143 lines
3.3 KiB
Go
|
// Copyright (C) 2018 Storj Labs, Inc.
|
||
|
// See LICENSE for copying information.
|
||
|
|
||
|
package server
|
||
|
|
||
|
import (
|
||
|
"fmt"
|
||
|
"io"
|
||
|
"log"
|
||
|
"os"
|
||
|
|
||
|
"golang.org/x/net/context"
|
||
|
|
||
|
"storj.io/storj/pkg/piecestore"
|
||
|
"storj.io/storj/pkg/piecestore/rpc/server/ttl"
|
||
|
pb "storj.io/storj/protos/piecestore"
|
||
|
)
|
||
|
|
||
|
// OK - Success!
|
||
|
const OK = "OK"
|
||
|
|
||
|
// Server -- GRPC server meta data used in route calls
|
||
|
type Server struct {
|
||
|
PieceStoreDir string
|
||
|
DB *ttl.TTL
|
||
|
}
|
||
|
|
||
|
// Store -- Store incoming data using piecestore
|
||
|
func (s *Server) Store(stream pb.PieceStoreRoutes_StoreServer) error {
|
||
|
log.Println("Storing data...")
|
||
|
|
||
|
// Receive initial meta data about what's being stored
|
||
|
piece, err := stream.Recv()
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// If we put in the database first then that checks if the data already exists
|
||
|
if err := s.DB.AddTTLToDB(piece.Id, piece.Ttl); err != nil {
|
||
|
log.Println(err)
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// Initialize file for storing data
|
||
|
storeFile, err := pstore.StoreWriter(piece.Id, piece.Size, piece.StoreOffset, s.PieceStoreDir)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
defer storeFile.Close()
|
||
|
|
||
|
reader := &StreamReader{stream: stream}
|
||
|
total, err := io.Copy(storeFile, reader)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
if total < piece.Size {
|
||
|
return fmt.Errorf("Received %v bytes of total %v bytes", int64(total), piece.Size)
|
||
|
}
|
||
|
|
||
|
log.Println("Successfully stored data.")
|
||
|
|
||
|
return stream.SendAndClose(&pb.PieceStoreSummary{Message: OK, TotalReceived: int64(total)})
|
||
|
}
|
||
|
|
||
|
// Retrieve -- Retrieve data from piecestore and send to client
|
||
|
func (s *Server) Retrieve(pieceMeta *pb.PieceRetrieval, stream pb.PieceStoreRoutes_RetrieveServer) error {
|
||
|
log.Println("Retrieving data...")
|
||
|
|
||
|
path, err := pstore.PathByID(pieceMeta.Id, s.PieceStoreDir)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
fileInfo, err := os.Stat(path)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// Read the size specified
|
||
|
totalToRead := pieceMeta.Size
|
||
|
// Read the entire file if specified -1
|
||
|
if pieceMeta.Size <= -1 {
|
||
|
totalToRead = fileInfo.Size()
|
||
|
}
|
||
|
|
||
|
storeFile, err := pstore.RetrieveReader(pieceMeta.Id, totalToRead, pieceMeta.StoreOffset, s.PieceStoreDir)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
defer storeFile.Close()
|
||
|
|
||
|
writer := &StreamWriter{stream: stream}
|
||
|
_, err = io.Copy(writer, storeFile)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
log.Println("Successfully retrieved data.")
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// Piece -- Send meta data about a stored by by Id
|
||
|
func (s *Server) Piece(ctx context.Context, in *pb.PieceId) (*pb.PieceSummary, error) {
|
||
|
log.Println("Getting Meta data...")
|
||
|
|
||
|
path, err := pstore.PathByID(in.Id, s.PieceStoreDir)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
fileInfo, err := os.Stat(path)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
// Read database to calculate expiration
|
||
|
ttl, err := s.DB.GetTTLByID(in.Id)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
log.Println("Meta data retrieved.")
|
||
|
return &pb.PieceSummary{Id: in.Id, Size: fileInfo.Size(), Expiration: ttl}, nil
|
||
|
}
|
||
|
|
||
|
// Delete -- Delete data by Id from piecestore
|
||
|
func (s *Server) Delete(ctx context.Context, in *pb.PieceDelete) (*pb.PieceDeleteSummary, error) {
|
||
|
log.Println("Deleting data...")
|
||
|
|
||
|
if err := pstore.Delete(in.Id, s.PieceStoreDir); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
if err := s.DB.DeleteTTLByID(in.Id); err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
log.Println("Successfully deleted data.")
|
||
|
return &pb.PieceDeleteSummary{Message: OK}, nil
|
||
|
}
|