aws_upload.go 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. package main
  2. // copied from https://github.com/apoorvam/aws-s3-multipart-upload
  3. import (
  4. "bytes"
  5. "flag"
  6. "fmt"
  7. "net/http"
  8. "os"
  9. "github.com/aws/aws-sdk-go/aws"
  10. "github.com/aws/aws-sdk-go/aws/awserr"
  11. "github.com/aws/aws-sdk-go/aws/credentials"
  12. "github.com/aws/aws-sdk-go/aws/session"
  13. "github.com/aws/aws-sdk-go/service/s3"
  14. )
  15. const (
  16. maxPartSize = int64(5 * 1024 * 1024)
  17. maxRetries = 3
  18. awsAccessKeyID = "Your access key"
  19. awsSecretAccessKey = "Your secret key"
  20. awsBucketRegion = "S3 bucket region"
  21. awsBucketName = "newBucket"
  22. )
  23. var (
  24. filename = flag.String("f", "", "the file name")
  25. )
  26. func main() {
  27. flag.Parse()
  28. creds := credentials.NewStaticCredentials(awsAccessKeyID, awsSecretAccessKey, "")
  29. _, err := creds.Get()
  30. if err != nil {
  31. fmt.Printf("bad credentials: %s", err)
  32. }
  33. cfg := aws.NewConfig().WithRegion(awsBucketRegion).WithCredentials(creds).WithDisableSSL(true).WithEndpoint("localhost:8333")
  34. svc := s3.New(session.New(), cfg)
  35. file, err := os.Open(*filename)
  36. if err != nil {
  37. fmt.Printf("err opening file: %s", err)
  38. return
  39. }
  40. defer file.Close()
  41. fileInfo, _ := file.Stat()
  42. size := fileInfo.Size()
  43. buffer := make([]byte, size)
  44. fileType := http.DetectContentType(buffer)
  45. file.Read(buffer)
  46. path := "/media/" + file.Name()
  47. input := &s3.CreateMultipartUploadInput{
  48. Bucket: aws.String(awsBucketName),
  49. Key: aws.String(path),
  50. ContentType: aws.String(fileType),
  51. }
  52. resp, err := svc.CreateMultipartUpload(input)
  53. if err != nil {
  54. fmt.Println(err.Error())
  55. return
  56. }
  57. fmt.Println("Created multipart upload request")
  58. var curr, partLength int64
  59. var remaining = size
  60. var completedParts []*s3.CompletedPart
  61. partNumber := 1
  62. for curr = 0; remaining != 0; curr += partLength {
  63. if remaining < maxPartSize {
  64. partLength = remaining
  65. } else {
  66. partLength = maxPartSize
  67. }
  68. completedPart, err := uploadPart(svc, resp, buffer[curr:curr+partLength], partNumber)
  69. if err != nil {
  70. fmt.Println(err.Error())
  71. err := abortMultipartUpload(svc, resp)
  72. if err != nil {
  73. fmt.Println(err.Error())
  74. }
  75. return
  76. }
  77. remaining -= partLength
  78. partNumber++
  79. completedParts = append(completedParts, completedPart)
  80. }
  81. // list parts
  82. parts, err := svc.ListParts(&s3.ListPartsInput{
  83. Bucket: input.Bucket,
  84. Key: input.Key,
  85. MaxParts: nil,
  86. PartNumberMarker: nil,
  87. RequestPayer: nil,
  88. UploadId: resp.UploadId,
  89. })
  90. if err != nil {
  91. fmt.Println(err.Error())
  92. return
  93. }
  94. fmt.Printf("list parts: %d\n", len(parts.Parts))
  95. for i, part := range parts.Parts {
  96. fmt.Printf("part %d: %v\n", i, part)
  97. }
  98. completeResponse, err := completeMultipartUpload(svc, resp, completedParts)
  99. if err != nil {
  100. fmt.Println(err.Error())
  101. return
  102. }
  103. fmt.Printf("Successfully uploaded file: %s\n", completeResponse.String())
  104. }
  105. func completeMultipartUpload(svc *s3.S3, resp *s3.CreateMultipartUploadOutput, completedParts []*s3.CompletedPart) (*s3.CompleteMultipartUploadOutput, error) {
  106. completeInput := &s3.CompleteMultipartUploadInput{
  107. Bucket: resp.Bucket,
  108. Key: resp.Key,
  109. UploadId: resp.UploadId,
  110. MultipartUpload: &s3.CompletedMultipartUpload{
  111. Parts: completedParts,
  112. },
  113. }
  114. return svc.CompleteMultipartUpload(completeInput)
  115. }
  116. func uploadPart(svc *s3.S3, resp *s3.CreateMultipartUploadOutput, fileBytes []byte, partNumber int) (*s3.CompletedPart, error) {
  117. tryNum := 1
  118. partInput := &s3.UploadPartInput{
  119. Body: bytes.NewReader(fileBytes),
  120. Bucket: resp.Bucket,
  121. Key: resp.Key,
  122. PartNumber: aws.Int64(int64(partNumber)),
  123. UploadId: resp.UploadId,
  124. ContentLength: aws.Int64(int64(len(fileBytes))),
  125. }
  126. for tryNum <= maxRetries {
  127. uploadResult, err := svc.UploadPart(partInput)
  128. if err != nil {
  129. if tryNum == maxRetries {
  130. if aerr, ok := err.(awserr.Error); ok {
  131. return nil, aerr
  132. }
  133. return nil, err
  134. }
  135. fmt.Printf("Retrying to upload part #%v\n", partNumber)
  136. tryNum++
  137. } else {
  138. fmt.Printf("Uploaded part #%v\n", partNumber)
  139. return &s3.CompletedPart{
  140. ETag: uploadResult.ETag,
  141. PartNumber: aws.Int64(int64(partNumber)),
  142. }, nil
  143. }
  144. }
  145. return nil, nil
  146. }
  147. func abortMultipartUpload(svc *s3.S3, resp *s3.CreateMultipartUploadOutput) error {
  148. fmt.Println("Aborting multipart upload for UploadId#" + *resp.UploadId)
  149. abortInput := &s3.AbortMultipartUploadInput{
  150. Bucket: resp.Bucket,
  151. Key: resp.Key,
  152. UploadId: resp.UploadId,
  153. }
  154. _, err := svc.AbortMultipartUpload(abortInput)
  155. return err
  156. }