gifs/api
2014-08-27
Child:b5d88d57d587
gifs/api/upload.go
Spike out functionality and tests. Create our interfaces around storing data and retrieving it. Create an in-memory implementation of our interfaces, for testing and rapid dev purposes. Begin sketching out what our unit tests look like. Create our Google Cloud Storage datastore implementation. Sketch out an idea for a usage collection process to keep track of which users are actually using stuff.
1 package api
3 import (
4 "crypto/sha1"
5 "encoding/hex"
6 "io"
7 "log"
9 "secondbit.org/uuid"
10 )
12 func Upload(collectionID uuid.ID, r io.Reader, c Context) (Item, int64, error) {
13 hashReader, hashWriter := io.Pipe()
14 uploadReader, uploadWriter := io.Pipe()
15 m := io.MultiWriter(hashWriter, uploadWriter)
17 hashChan := make(chan string)
18 hashError := make(chan error)
19 hashDone := make(chan struct{})
21 cpDone := make(chan struct{})
22 cpErr := make(chan error)
24 uploadDone := make(chan struct{})
25 uploadErr := make(chan error)
27 n := make(chan int64)
29 var bytesWritten int64
30 tmp := uuid.NewID().String()
31 tmp = "tmp/" + tmp
33 go hash(hashReader, hashChan, hashError, hashDone)
34 go asyncCopy(m, r, n, cpErr, cpDone)
35 if c.Storage != nil {
36 go c.Storage.Upload(c.Bucket, tmp, uploadReader, uploadErr, uploadDone)
37 }
39 select {
40 case err := <-hashError:
41 if err != nil {
42 hashWriter.CloseWithError(err)
43 uploadWriter.CloseWithError(err)
44 return Item{}, 0, err
45 } else {
46 hashWriter.Close()
47 uploadWriter.Close()
48 }
49 case err := <-cpErr:
50 if err != nil {
51 hashWriter.CloseWithError(err)
52 uploadWriter.CloseWithError(err)
53 return Item{}, 0, err
54 } else {
55 hashWriter.Close()
56 uploadWriter.Close()
57 }
58 case err := <-uploadErr:
59 if err != nil {
60 hashWriter.CloseWithError(err)
61 uploadWriter.CloseWithError(err)
62 return Item{}, 0, err
63 } else {
64 hashWriter.Close()
65 uploadWriter.Close()
66 }
67 case <-cpDone:
68 hashWriter.Close()
69 uploadWriter.Close()
70 }
71 <-cpDone
72 <-hashDone
73 if c.Storage != nil {
74 <-uploadDone
75 }
76 bytesWritten = <-n
77 finalLocation := <-hashChan
78 if c.Storage != nil {
79 err := c.Storage.Move(c.Bucket, tmp, c.Bucket, finalLocation)
80 if err != nil {
81 return Item{}, 0, err
82 }
83 }
84 return Item{
85 Blob: finalLocation,
86 Bucket: c.Bucket,
87 CollectionID: collectionID,
88 }, bytesWritten, nil
89 }
91 func hash(r io.Reader, resp chan string, errs chan error, done chan struct{}) {
92 if resp != nil {
93 defer close(resp)
94 }
95 h := sha1.New()
96 go asyncCopy(h, r, nil, errs, done)
97 <-done
98 resp <- hex.EncodeToString(h.Sum(nil))
99 }
101 func del(bucket, tmp string, c Context) {
102 err := c.Storage.Delete(bucket, tmp)
103 if err != nil {
104 log.Printf("Error deleting temporary upload %s in %s: %s\n", tmp, bucket, err)
105 }
106 }
108 func asyncCopy(dst io.Writer, src io.Reader, n chan int64, errs chan error, done chan struct{}) {
109 if errs != nil {
110 defer close(errs)
111 }
112 if done != nil {
113 defer close(done)
114 }
115 written, err := io.Copy(dst, src)
116 if errs != nil && err != nil {
117 errs <- err
118 }
119 if n != nil {
120 go func(n chan int64, written int64) {
121 n <- written
122 }(n, written)
123 }
124 }