media.go 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. // Copyright 2016 The Go Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package gensupport
  5. import (
  6. "fmt"
  7. "io"
  8. "io/ioutil"
  9. "mime/multipart"
  10. "net/http"
  11. "net/textproto"
  12. "google.golang.org/api/googleapi"
  13. )
  14. const sniffBuffSize = 512
  15. func newContentSniffer(r io.Reader) *contentSniffer {
  16. return &contentSniffer{r: r}
  17. }
  18. // contentSniffer wraps a Reader, and reports the content type determined by sniffing up to 512 bytes from the Reader.
  19. type contentSniffer struct {
  20. r io.Reader
  21. start []byte // buffer for the sniffed bytes.
  22. err error // set to any error encountered while reading bytes to be sniffed.
  23. ctype string // set on first sniff.
  24. sniffed bool // set to true on first sniff.
  25. }
  26. func (cs *contentSniffer) Read(p []byte) (n int, err error) {
  27. // Ensure that the content type is sniffed before any data is consumed from Reader.
  28. _, _ = cs.ContentType()
  29. if len(cs.start) > 0 {
  30. n := copy(p, cs.start)
  31. cs.start = cs.start[n:]
  32. return n, nil
  33. }
  34. // We may have read some bytes into start while sniffing, even if the read ended in an error.
  35. // We should first return those bytes, then the error.
  36. if cs.err != nil {
  37. return 0, cs.err
  38. }
  39. // Now we have handled all bytes that were buffered while sniffing. Now just delegate to the underlying reader.
  40. return cs.r.Read(p)
  41. }
  42. // ContentType returns the sniffed content type, and whether the content type was succesfully sniffed.
  43. func (cs *contentSniffer) ContentType() (string, bool) {
  44. if cs.sniffed {
  45. return cs.ctype, cs.ctype != ""
  46. }
  47. cs.sniffed = true
  48. // If ReadAll hits EOF, it returns err==nil.
  49. cs.start, cs.err = ioutil.ReadAll(io.LimitReader(cs.r, sniffBuffSize))
  50. // Don't try to detect the content type based on possibly incomplete data.
  51. if cs.err != nil {
  52. return "", false
  53. }
  54. cs.ctype = http.DetectContentType(cs.start)
  55. return cs.ctype, true
  56. }
  57. // DetermineContentType determines the content type of the supplied reader.
  58. // If the content type is already known, it can be specified via ctype.
  59. // Otherwise, the content of media will be sniffed to determine the content type.
  60. // If media implements googleapi.ContentTyper (deprecated), this will be used
  61. // instead of sniffing the content.
  62. // After calling DetectContentType the caller must not perform further reads on
  63. // media, but rather read from the Reader that is returned.
  64. func DetermineContentType(media io.Reader, ctype string) (io.Reader, string) {
  65. // Note: callers could avoid calling DetectContentType if ctype != "",
  66. // but doing the check inside this function reduces the amount of
  67. // generated code.
  68. if ctype != "" {
  69. return media, ctype
  70. }
  71. // For backwards compatability, allow clients to set content
  72. // type by providing a ContentTyper for media.
  73. if typer, ok := media.(googleapi.ContentTyper); ok {
  74. return media, typer.ContentType()
  75. }
  76. sniffer := newContentSniffer(media)
  77. if ctype, ok := sniffer.ContentType(); ok {
  78. return sniffer, ctype
  79. }
  80. // If content type could not be sniffed, reads from sniffer will eventually fail with an error.
  81. return sniffer, ""
  82. }
  83. type typeReader struct {
  84. io.Reader
  85. typ string
  86. }
  87. // multipartReader combines the contents of multiple readers to creat a multipart/related HTTP body.
  88. // Close must be called if reads from the multipartReader are abandoned before reaching EOF.
  89. type multipartReader struct {
  90. pr *io.PipeReader
  91. pipeOpen bool
  92. ctype string
  93. }
  94. func newMultipartReader(parts []typeReader) *multipartReader {
  95. mp := &multipartReader{pipeOpen: true}
  96. var pw *io.PipeWriter
  97. mp.pr, pw = io.Pipe()
  98. mpw := multipart.NewWriter(pw)
  99. mp.ctype = "multipart/related; boundary=" + mpw.Boundary()
  100. go func() {
  101. for _, part := range parts {
  102. w, err := mpw.CreatePart(typeHeader(part.typ))
  103. if err != nil {
  104. mpw.Close()
  105. pw.CloseWithError(fmt.Errorf("googleapi: CreatePart failed: %v", err))
  106. return
  107. }
  108. _, err = io.Copy(w, part.Reader)
  109. if err != nil {
  110. mpw.Close()
  111. pw.CloseWithError(fmt.Errorf("googleapi: Copy failed: %v", err))
  112. return
  113. }
  114. }
  115. mpw.Close()
  116. pw.Close()
  117. }()
  118. return mp
  119. }
  120. func (mp *multipartReader) Read(data []byte) (n int, err error) {
  121. return mp.pr.Read(data)
  122. }
  123. func (mp *multipartReader) Close() error {
  124. if !mp.pipeOpen {
  125. return nil
  126. }
  127. mp.pipeOpen = false
  128. return mp.pr.Close()
  129. }
  130. // CombineBodyMedia combines a json body with media content to create a multipart/related HTTP body.
  131. // It returns a ReadCloser containing the combined body, and the overall "multipart/related" content type, with random boundary.
  132. //
  133. // The caller must call Close on the returned ReadCloser if reads are abandoned before reaching EOF.
  134. func CombineBodyMedia(body io.Reader, bodyContentType string, media io.Reader, mediaContentType string) (io.ReadCloser, string) {
  135. mp := newMultipartReader([]typeReader{
  136. {body, bodyContentType},
  137. {media, mediaContentType},
  138. })
  139. return mp, mp.ctype
  140. }
  141. func typeHeader(contentType string) textproto.MIMEHeader {
  142. h := make(textproto.MIMEHeader)
  143. if contentType != "" {
  144. h.Set("Content-Type", contentType)
  145. }
  146. return h
  147. }
  148. // PrepareUpload determines whether the data in the supplied reader should be
  149. // uploaded in a single request, or in sequential chunks.
  150. // chunkSize is the size of the chunk that media should be split into.
  151. //
  152. // If chunkSize is zero, media is returned as the first value, and the other
  153. // two return values are nil, true.
  154. //
  155. // Otherwise, a MediaBuffer is returned, along with a bool indicating whether the
  156. // contents of media fit in a single chunk.
  157. //
  158. // After PrepareUpload has been called, media should no longer be used: the
  159. // media content should be accessed via one of the return values.
  160. func PrepareUpload(media io.Reader, chunkSize int) (r io.Reader, mb *MediaBuffer, singleChunk bool) {
  161. if chunkSize == 0 { // do not chunk
  162. return media, nil, true
  163. }
  164. mb = NewMediaBuffer(media, chunkSize)
  165. _, _, _, err := mb.Chunk()
  166. // If err is io.EOF, we can upload this in a single request. Otherwise, err is
  167. // either nil or a non-EOF error. If it is the latter, then the next call to
  168. // mb.Chunk will return the same error. Returning a MediaBuffer ensures that this
  169. // error will be handled at some point.
  170. return nil, mb, err == io.EOF
  171. }
  172. // MediaInfo holds information for media uploads. It is intended for use by generated
  173. // code only.
  174. type MediaInfo struct {
  175. // At most one of Media and MediaBuffer will be set.
  176. media io.Reader
  177. buffer *MediaBuffer
  178. singleChunk bool
  179. mType string
  180. size int64 // mediaSize, if known. Used only for calls to progressUpdater_.
  181. progressUpdater googleapi.ProgressUpdater
  182. }
  183. // NewInfoFromMedia should be invoked from the Media method of a call. It returns a
  184. // MediaInfo populated with chunk size and content type, and a reader or MediaBuffer
  185. // if needed.
  186. func NewInfoFromMedia(r io.Reader, options []googleapi.MediaOption) *MediaInfo {
  187. mi := &MediaInfo{}
  188. opts := googleapi.ProcessMediaOptions(options)
  189. if !opts.ForceEmptyContentType {
  190. r, mi.mType = DetermineContentType(r, opts.ContentType)
  191. }
  192. mi.media, mi.buffer, mi.singleChunk = PrepareUpload(r, opts.ChunkSize)
  193. return mi
  194. }
  195. // NewInfoFromResumableMedia should be invoked from the ResumableMedia method of a
  196. // call. It returns a MediaInfo using the given reader, size and media type.
  197. func NewInfoFromResumableMedia(r io.ReaderAt, size int64, mediaType string) *MediaInfo {
  198. rdr := ReaderAtToReader(r, size)
  199. rdr, mType := DetermineContentType(rdr, mediaType)
  200. return &MediaInfo{
  201. size: size,
  202. mType: mType,
  203. buffer: NewMediaBuffer(rdr, googleapi.DefaultUploadChunkSize),
  204. media: nil,
  205. singleChunk: false,
  206. }
  207. }
  208. func (mi *MediaInfo) SetProgressUpdater(pu googleapi.ProgressUpdater) {
  209. if mi != nil {
  210. mi.progressUpdater = pu
  211. }
  212. }
  213. // UploadType determines the type of upload: a single request, or a resumable
  214. // series of requests.
  215. func (mi *MediaInfo) UploadType() string {
  216. if mi.singleChunk {
  217. return "multipart"
  218. }
  219. return "resumable"
  220. }
  221. // UploadRequest sets up an HTTP request for media upload. It adds headers
  222. // as necessary, and returns a replacement for the body.
  223. func (mi *MediaInfo) UploadRequest(reqHeaders http.Header, body io.Reader) (newBody io.Reader, cleanup func()) {
  224. cleanup = func() {}
  225. if mi == nil {
  226. return body, cleanup
  227. }
  228. var media io.Reader
  229. if mi.media != nil {
  230. // This only happens when the caller has turned off chunking. In that
  231. // case, we write all of media in a single non-retryable request.
  232. media = mi.media
  233. } else if mi.singleChunk {
  234. // The data fits in a single chunk, which has now been read into the MediaBuffer.
  235. // We obtain that chunk so we can write it in a single request. The request can
  236. // be retried because the data is stored in the MediaBuffer.
  237. media, _, _, _ = mi.buffer.Chunk()
  238. }
  239. if media != nil {
  240. combined, ctype := CombineBodyMedia(body, "application/json", media, mi.mType)
  241. cleanup = func() { combined.Close() }
  242. reqHeaders.Set("Content-Type", ctype)
  243. body = combined
  244. }
  245. if mi.buffer != nil && mi.mType != "" && !mi.singleChunk {
  246. reqHeaders.Set("X-Upload-Content-Type", mi.mType)
  247. }
  248. return body, cleanup
  249. }
  250. // ResumableUpload returns an appropriately configured ResumableUpload value if the
  251. // upload is resumable, or nil otherwise.
  252. func (mi *MediaInfo) ResumableUpload(locURI string) *ResumableUpload {
  253. if mi == nil || mi.singleChunk {
  254. return nil
  255. }
  256. return &ResumableUpload{
  257. URI: locURI,
  258. Media: mi.buffer,
  259. MediaType: mi.mType,
  260. Callback: func(curr int64) {
  261. if mi.progressUpdater != nil {
  262. mi.progressUpdater(curr, mi.size)
  263. }
  264. },
  265. }
  266. }