vendor/github.com/aliyun/aliyun-oss-go-sdk/oss/progress.go

   1 package oss
   2
   3 import "io"
   4
   5 // ProgressEventType defines transfer progress event type
   6 type ProgressEventType int
   7
   8 const (
   9         // TransferStartedEvent transfer started, set TotalBytes
  10         TransferStartedEvent ProgressEventType = 1 + iota
  11         // TransferDataEvent transfer data, set ConsumedBytes anmd TotalBytes
  12         TransferDataEvent
  13         // TransferCompletedEvent transfer completed
  14         TransferCompletedEvent
  15         // TransferFailedEvent transfer encounters an error
  16         TransferFailedEvent
  17 )
  18
  19 // ProgressEvent defines progress event
  20 type ProgressEvent struct {
  21         ConsumedBytes int64
  22         TotalBytes    int64
  23         RwBytes       int64
  24         EventType     ProgressEventType
  25 }
  26
  27 // ProgressListener listens progress change
  28 type ProgressListener interface {
  29         ProgressChanged(event *ProgressEvent)
  30 }
  31
  32 // -------------------- Private --------------------
  33
  34 func newProgressEvent(eventType ProgressEventType, consumed, total int64, rwBytes int64) *ProgressEvent {
  35         return &ProgressEvent{
  36                 ConsumedBytes: consumed,
  37                 TotalBytes:    total,
  38                 RwBytes:       rwBytes,
  39                 EventType:     eventType}
  40 }
  41
  42 // publishProgress
  43 func publishProgress(listener ProgressListener, event *ProgressEvent) {
  44         if listener != nil && event != nil {
  45                 listener.ProgressChanged(event)
  46         }
  47 }
  48
  49 type readerTracker struct {
  50         completedBytes int64
  51 }
  52
  53 type teeReader struct {
  54         reader        io.Reader
  55         writer        io.Writer
  56         listener      ProgressListener
  57         consumedBytes int64
  58         totalBytes    int64
  59         tracker       *readerTracker
  60 }
  61
  62 // TeeReader returns a Reader that writes to w what it reads from r.
  63 // All reads from r performed through it are matched with
  64 // corresponding writes to w.  There is no internal buffering -
  65 // the write must complete before the read completes.
  66 // Any error encountered while writing is reported as a read error.
  67 func TeeReader(reader io.Reader, writer io.Writer, totalBytes int64, listener ProgressListener, tracker *readerTracker) io.ReadCloser {
  68         return &teeReader{
  69                 reader:        reader,
  70                 writer:        writer,
  71                 listener:      listener,
  72                 consumedBytes: 0,
  73                 totalBytes:    totalBytes,
  74                 tracker:       tracker,
  75         }
  76 }
  77
  78 func (t *teeReader) Read(p []byte) (n int, err error) {
  79         n, err = t.reader.Read(p)
  80
  81         // Read encountered error
  82         if err != nil && err != io.EOF {
  83                 event := newProgressEvent(TransferFailedEvent, t.consumedBytes, t.totalBytes, 0)
  84                 publishProgress(t.listener, event)
  85         }
  86
  87         if n > 0 {
  88                 t.consumedBytes += int64(n)
  89                 // CRC
  90                 if t.writer != nil {
  91                         if n, err := t.writer.Write(p[:n]); err != nil {
  92                                 return n, err
  93                         }
  94                 }
  95                 // Progress
  96                 if t.listener != nil {
  97                         event := newProgressEvent(TransferDataEvent, t.consumedBytes, t.totalBytes, int64(n))
  98                         publishProgress(t.listener, event)
  99                 }
 100                 // Track
 101                 if t.tracker != nil {
 102                         t.tracker.completedBytes = t.consumedBytes
 103                 }
 104         }
 105
 106         return
 107 }
 108
 109 func (t *teeReader) Close() error {
 110         if rc, ok := t.reader.(io.ReadCloser); ok {
 111                 return rc.Close()
 112         }
 113         return nil
 114 }