You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

392 lines
8.4 KiB

9 years ago
10 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
  1. package util
  2. import (
  3. "bytes"
  4. "compress/gzip"
  5. "encoding/json"
  6. "errors"
  7. "fmt"
  8. "io"
  9. "io/ioutil"
  10. "net/http"
  11. "net/url"
  12. "strings"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. )
  15. var (
  16. client *http.Client
  17. Transport *http.Transport
  18. )
  19. func init() {
  20. Transport = &http.Transport{
  21. MaxIdleConnsPerHost: 1024,
  22. }
  23. client = &http.Client{
  24. Transport: Transport,
  25. }
  26. }
  27. func PostBytes(url string, body []byte) ([]byte, error) {
  28. r, err := client.Post(url, "", bytes.NewReader(body))
  29. if err != nil {
  30. return nil, fmt.Errorf("Post to %s: %v", url, err)
  31. }
  32. defer r.Body.Close()
  33. b, err := ioutil.ReadAll(r.Body)
  34. if err != nil {
  35. return nil, fmt.Errorf("Read response body: %v", err)
  36. }
  37. if r.StatusCode >= 400 {
  38. return nil, fmt.Errorf("%s: %s", url, r.Status)
  39. }
  40. return b, nil
  41. }
  42. func Post(url string, values url.Values) ([]byte, error) {
  43. r, err := client.PostForm(url, values)
  44. if err != nil {
  45. return nil, err
  46. }
  47. defer r.Body.Close()
  48. b, err := ioutil.ReadAll(r.Body)
  49. if r.StatusCode >= 400 {
  50. if err != nil {
  51. return nil, fmt.Errorf("%s: %d - %s", url, r.StatusCode, string(b))
  52. } else {
  53. return nil, fmt.Errorf("%s: %s", url, r.Status)
  54. }
  55. }
  56. if err != nil {
  57. return nil, err
  58. }
  59. return b, nil
  60. }
  61. // github.com/chrislusf/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go
  62. // may need increasing http.Client.Timeout
  63. func Get(url string) ([]byte, error) {
  64. request, err := http.NewRequest("GET", url, nil)
  65. request.Header.Add("Accept-Encoding", "gzip")
  66. response, err := client.Do(request)
  67. if err != nil {
  68. return nil, err
  69. }
  70. defer response.Body.Close()
  71. var reader io.ReadCloser
  72. switch response.Header.Get("Content-Encoding") {
  73. case "gzip":
  74. reader, err = gzip.NewReader(response.Body)
  75. defer reader.Close()
  76. default:
  77. reader = response.Body
  78. }
  79. b, err := ioutil.ReadAll(reader)
  80. if response.StatusCode >= 400 {
  81. return nil, fmt.Errorf("%s: %s", url, response.Status)
  82. }
  83. if err != nil {
  84. return nil, err
  85. }
  86. return b, nil
  87. }
  88. func Head(url string) (http.Header, error) {
  89. r, err := client.Head(url)
  90. if err != nil {
  91. return nil, err
  92. }
  93. defer CloseResponse(r)
  94. if r.StatusCode >= 400 {
  95. return nil, fmt.Errorf("%s: %s", url, r.Status)
  96. }
  97. return r.Header, nil
  98. }
  99. func Delete(url string, jwt string) error {
  100. req, err := http.NewRequest("DELETE", url, nil)
  101. if jwt != "" {
  102. req.Header.Set("Authorization", "BEARER "+string(jwt))
  103. }
  104. if err != nil {
  105. return err
  106. }
  107. resp, e := client.Do(req)
  108. if e != nil {
  109. return e
  110. }
  111. defer resp.Body.Close()
  112. body, err := ioutil.ReadAll(resp.Body)
  113. if err != nil {
  114. return err
  115. }
  116. switch resp.StatusCode {
  117. case http.StatusNotFound, http.StatusAccepted, http.StatusOK:
  118. return nil
  119. }
  120. m := make(map[string]interface{})
  121. if e := json.Unmarshal(body, &m); e == nil {
  122. if s, ok := m["error"].(string); ok {
  123. return errors.New(s)
  124. }
  125. }
  126. return errors.New(string(body))
  127. }
  128. func GetBufferStream(url string, values url.Values, allocatedBytes []byte, eachBuffer func([]byte)) error {
  129. r, err := client.PostForm(url, values)
  130. if err != nil {
  131. return err
  132. }
  133. defer CloseResponse(r)
  134. if r.StatusCode != 200 {
  135. return fmt.Errorf("%s: %s", url, r.Status)
  136. }
  137. for {
  138. n, err := r.Body.Read(allocatedBytes)
  139. if n > 0 {
  140. eachBuffer(allocatedBytes[:n])
  141. }
  142. if err != nil {
  143. if err == io.EOF {
  144. return nil
  145. }
  146. return err
  147. }
  148. }
  149. }
  150. func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) error {
  151. r, err := client.PostForm(url, values)
  152. if err != nil {
  153. return err
  154. }
  155. defer CloseResponse(r)
  156. if r.StatusCode != 200 {
  157. return fmt.Errorf("%s: %s", url, r.Status)
  158. }
  159. return readFn(r.Body)
  160. }
  161. func DownloadFile(fileUrl string) (filename string, header http.Header, rc io.ReadCloser, e error) {
  162. response, err := client.Get(fileUrl)
  163. if err != nil {
  164. return "", nil, nil, err
  165. }
  166. header = response.Header
  167. contentDisposition := response.Header["Content-Disposition"]
  168. if len(contentDisposition) > 0 {
  169. idx := strings.Index(contentDisposition[0], "filename=")
  170. if idx != -1 {
  171. filename = contentDisposition[0][idx+len("filename="):]
  172. filename = strings.Trim(filename, "\"")
  173. }
  174. }
  175. rc = response.Body
  176. return
  177. }
  178. func Do(req *http.Request) (resp *http.Response, err error) {
  179. return client.Do(req)
  180. }
  181. func NormalizeUrl(url string) string {
  182. if strings.HasPrefix(url, "http://") || strings.HasPrefix(url, "https://") {
  183. return url
  184. }
  185. return "http://" + url
  186. }
  187. func ReadUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, buf []byte) (int64, error) {
  188. if cipherKey != nil {
  189. var n int
  190. err := readEncryptedUrl(fileUrl, cipherKey, isContentCompressed, isFullChunk, offset, size, func(data []byte) {
  191. n = copy(buf, data)
  192. })
  193. return int64(n), err
  194. }
  195. req, err := http.NewRequest("GET", fileUrl, nil)
  196. if err != nil {
  197. return 0, err
  198. }
  199. if !isFullChunk {
  200. req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)-1))
  201. } else {
  202. req.Header.Set("Accept-Encoding", "gzip")
  203. }
  204. r, err := client.Do(req)
  205. if err != nil {
  206. return 0, err
  207. }
  208. defer r.Body.Close()
  209. if r.StatusCode >= 400 {
  210. return 0, fmt.Errorf("%s: %s", fileUrl, r.Status)
  211. }
  212. var reader io.ReadCloser
  213. contentEncoding := r.Header.Get("Content-Encoding")
  214. switch contentEncoding {
  215. case "gzip":
  216. reader, err = gzip.NewReader(r.Body)
  217. defer reader.Close()
  218. default:
  219. reader = r.Body
  220. }
  221. var (
  222. i, m int
  223. n int64
  224. )
  225. // refers to https://github.com/golang/go/blob/master/src/bytes/buffer.go#L199
  226. // commit id c170b14c2c1cfb2fd853a37add92a82fd6eb4318
  227. for {
  228. m, err = reader.Read(buf[i:])
  229. i += m
  230. n += int64(m)
  231. if err == io.EOF {
  232. return n, nil
  233. }
  234. if err != nil {
  235. return n, err
  236. }
  237. if n == int64(len(buf)) {
  238. break
  239. }
  240. }
  241. // drains the response body to avoid memory leak
  242. data, _ := ioutil.ReadAll(reader)
  243. if len(data) != 0 {
  244. glog.V(1).Infof("%s reader has remaining %d bytes", contentEncoding, len(data))
  245. }
  246. return n, err
  247. }
  248. func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) error {
  249. if cipherKey != nil {
  250. return readEncryptedUrl(fileUrl, cipherKey, isContentGzipped, isFullChunk, offset, size, fn)
  251. }
  252. req, err := http.NewRequest("GET", fileUrl, nil)
  253. if err != nil {
  254. return err
  255. }
  256. if isFullChunk {
  257. req.Header.Add("Accept-Encoding", "gzip")
  258. } else {
  259. req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)-1))
  260. }
  261. r, err := client.Do(req)
  262. if err != nil {
  263. return err
  264. }
  265. defer CloseResponse(r)
  266. if r.StatusCode >= 400 {
  267. return fmt.Errorf("%s: %s", fileUrl, r.Status)
  268. }
  269. var reader io.ReadCloser
  270. contentEncoding := r.Header.Get("Content-Encoding")
  271. switch contentEncoding {
  272. case "gzip":
  273. reader, err = gzip.NewReader(r.Body)
  274. defer reader.Close()
  275. default:
  276. reader = r.Body
  277. }
  278. var (
  279. m int
  280. )
  281. buf := make([]byte, 64*1024)
  282. for {
  283. m, err = reader.Read(buf)
  284. fn(buf[:m])
  285. if err == io.EOF {
  286. return nil
  287. }
  288. if err != nil {
  289. return err
  290. }
  291. }
  292. }
  293. func readEncryptedUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) error {
  294. encryptedData, err := Get(fileUrl)
  295. if err != nil {
  296. return fmt.Errorf("fetch %s: %v", fileUrl, err)
  297. }
  298. decryptedData, err := Decrypt(encryptedData, CipherKey(cipherKey))
  299. if err != nil {
  300. return fmt.Errorf("decrypt %s: %v", fileUrl, err)
  301. }
  302. if isContentCompressed {
  303. decryptedData, err = DecompressData(decryptedData)
  304. if err != nil {
  305. glog.V(0).Infof("unzip decrypt %s: %v", fileUrl, err)
  306. }
  307. }
  308. if len(decryptedData) < int(offset)+size {
  309. return fmt.Errorf("read decrypted %s size %d [%d, %d)", fileUrl, len(decryptedData), offset, int(offset)+size)
  310. }
  311. if isFullChunk {
  312. fn(decryptedData)
  313. } else {
  314. fn(decryptedData[int(offset) : int(offset)+size])
  315. }
  316. return nil
  317. }
  318. func ReadUrlAsReaderCloser(fileUrl string, rangeHeader string) (io.ReadCloser, error) {
  319. req, err := http.NewRequest("GET", fileUrl, nil)
  320. if err != nil {
  321. return nil, err
  322. }
  323. if rangeHeader != "" {
  324. req.Header.Add("Range", rangeHeader)
  325. } else {
  326. req.Header.Add("Accept-Encoding", "gzip")
  327. }
  328. r, err := client.Do(req)
  329. if err != nil {
  330. return nil, err
  331. }
  332. defer CloseResponse(r)
  333. if r.StatusCode >= 400 {
  334. return nil, fmt.Errorf("%s: %s", fileUrl, r.Status)
  335. }
  336. var reader io.ReadCloser
  337. contentEncoding := r.Header.Get("Content-Encoding")
  338. switch contentEncoding {
  339. case "gzip":
  340. reader, err = gzip.NewReader(r.Body)
  341. defer reader.Close()
  342. default:
  343. reader = r.Body
  344. }
  345. return reader, nil
  346. }
  347. func CloseResponse(resp *http.Response) {
  348. io.Copy(ioutil.Discard, resp.Body)
  349. resp.Body.Close()
  350. }