@@ -3,22 +3,38 @@ package data_store
33import (
44 "context"
55 "fmt"
6- "go.uber.org/zap"
76 "io"
87 "os"
9- "strings"
8+ "path/filepath"
9+ "time"
10+
11+ "go.uber.org/zap"
1012)
1113
1214type LocalClient struct {
13- DataPath string
15+ dataPath string
1416}
1517
16- func newLocalClient (config DataStoreConfig ) (LocalClient , error ) {
17- return LocalClient {DataPath : config .DataPath }, nil
18+ func newLocalClient (config DataStoreConfig ) (* LocalClient , error ) {
19+ return & LocalClient {
20+ dataPath : config .DataPath ,
21+ }, nil
1822}
1923
20- func (c LocalClient ) GetFile (object string , bucket string ) ([]byte , error ) {
21- targetObject := fmt .Sprintf ("%s/%s/%s" , c .DataPath , bucket , object )
24+ func (c * LocalClient ) GetDataPath () string {
25+ return c .dataPath
26+ }
27+
28+ func (c * LocalClient ) GetFile (object string , bucket string ) ([]byte , error ) {
29+ if len (bucket ) == 0 || len (object ) == 0 {
30+ zap .S ().Errorf ("Bucket or object are empty" )
31+ return nil , fmt .Errorf ("Bucket or object are empty" )
32+ }
33+
34+ start := time .Now ()
35+ defer elapsed (start , "[" + c .StorageType ()+ "] Get file" )
36+
37+ targetObject := fmt .Sprintf ("%s/%s/%s" , c .GetDataPath (), bucket , object )
2238 data , err := os .ReadFile (targetObject )
2339 if err != nil {
2440 zap .S ().Errorf ("err when getting object from store: %v" , err .Error ())
@@ -28,58 +44,82 @@ func (c LocalClient) GetFile(object string, bucket string) ([]byte, error) {
2844 return data , nil
2945}
3046
31- func (c LocalClient ) List (bucket string , prefix string ) ([]string , error ) {
32- var list []string
33- files , err := os .ReadDir (fmt .Sprintf ("%s/%s" , c .DataPath , bucket ))
47+ func (c * LocalClient ) List (bucket string , prefix string ) ([]string , error ) {
48+ return list (c , bucket , prefix )
49+ }
50+
51+ func (c * LocalClient ) ListChan (ctx context.Context , bucket string , prefix string ) (<- chan string , error ) {
52+ if len (bucket ) == 0 || len (prefix ) == 0 {
53+ zap .S ().Errorf ("Bucket or prefix are empty" )
54+ return nil , fmt .Errorf ("Bucket or prefix are empty" )
55+ }
56+
57+ start := time .Now ()
58+ defer elapsed (start , "[" + c .StorageType ()+ "] List channel files" )
59+
60+ list , err := filepath .Glob (fmt .Sprintf ("%s/%s/%s*" , c .GetDataPath (), bucket , prefix ))
3461 if err != nil {
3562 zap .S ().Errorf ("could not read directory '%s': %v" , bucket , err )
3663 return nil , err
3764 }
3865
39- for _ , file := range files {
40- fileName := file .Name ()
41- if strings .Contains (fileName , prefix ) {
42- list = append (list , fileName )
66+ outChan := make (chan string , 10 )
67+ go func (files []string ) {
68+ defer close (outChan )
69+
70+ for _ , f := range files {
71+ select {
72+ case <- ctx .Done ():
73+ return
74+ default :
75+ outChan <- filepath .Base (f )
76+ }
4377 }
44- }
78+ }( list )
4579
46- return list , nil
80+ return outChan , nil
4781}
4882
49- func (c LocalClient ) ListChan ( ctx context. Context , bucket string , prefix string ) ( <- chan string , error ) {
50- panic ( "not implemented" )
83+ func (c * LocalClient ) UploadFromFile ( name string , folder string ) error {
84+ return uploadFromFile ( c , name , folder )
5185}
5286
53- func (c LocalClient ) UploadFromFile (name string , dest string ) error {
54- file , err := os .Open (name )
55- if err != nil {
56- return err
57- }
58- defer file .Close ()
87+ func (c * LocalClient ) UploadFromBytes (data []byte , folder string , name string ) error {
88+ return uploadFromBytes (c , data , folder , name )
89+ }
5990
60- fileStat , err := file .Stat ()
61- if err != nil {
62- return err
91+ func (c * LocalClient ) UploadFromReader (data io.Reader , size int64 , folder string , name string ) error {
92+ if len (folder ) == 0 || len (name ) == 0 {
93+ zap .S ().Errorf ("Folder or name are empty" )
94+ return fmt .Errorf ("Folder or name are empty" )
6395 }
6496
65- if ! fileStat .Mode ().IsRegular () {
66- return fmt .Errorf ("%s is not a regular file" , name )
67- }
97+ start := time .Now ()
98+ defer elapsed (start , "[" + c .StorageType ()+ "] Upload from reader" )
6899
69- out , err := os .Create (fmt .Sprintf ("%s/%s/%s" , c .DataPath , dest , name ))
100+ destFolder := fmt .Sprintf ("%s/%s" , c .GetDataPath (), folder )
101+ if _ , err := os .Stat (destFolder ); err != nil {
102+ if ! os .IsNotExist (err ) {
103+ return err
104+ }
105+ if err := os .MkdirAll (destFolder , os .ModePerm ); err != nil {
106+ return err
107+ }
108+ }
109+ destUrl := fmt .Sprintf ("%s/%s" , destFolder , name )
110+ out , err := os .Create (destUrl )
70111 if err != nil {
71112 return err
72113 }
73114 defer out .Close ()
74115
75- _ , err = io .Copy (out , file )
76- return err
77- }
116+ _ , err = io .Copy (out , data )
117+
118+ zap . S (). Debugf ( "[%s] Operation: upload, Source: %s, Destination: %s, Size: %d" , c . StorageType (), name , destUrl , size )
78119
79- func (c LocalClient ) UploadFromBytes (data []byte , destFolder string , destName string ) error {
80- panic ("not implemented" )
120+ return err
81121}
82122
83- func (c LocalClient ) StorageType () string {
123+ func (c * LocalClient ) StorageType () string {
84124 return LocalStorage
85125}
0 commit comments