Code snippets for the blog post "Asynchronously Split an io.Reader in Go" (http://rodaine.com/2015/04/async-split-io-reader-in-golang/)

bytesreader.go

func handleUpload(u io.Reader) (err error) {
  // capture all bytes from upload
  b, err := ioutil.ReadAll(u)
  if err != nil {
    return
  }

  // wrap the bytes in a ReadSeeker
  r := bytes.NewReader(b)

  // process the meta data
  err = processMetaData(r)
  if err != nil {
    return
  }

  // rewind the reader back to the start
  r.Seek(0, 0)

  // upload the data
  err = uploadFile(r)
  if err != nil {
    return
  }

  return nil
}

file.go

func handleUpload(u io.Reader) (err error) {
  // create a temporary file for the upload
  f, err := ioutil.TempFile("", "upload")
  if err != nil {
    return
  }

  // destroy the file once done
  defer func() {
    n := f.Name()
    f.Close()
    os.Remove(n)
  }()

  // transfer the bytes to the file
  _, err = io.Copy(f, u)
  if err != nil {
    return
  }

  // rewind the file
  f.Seek(0, 0)

  // process the meta data
  err = processMetaData(f)
  if err != nil {
    return
  }

  // rewind the file again
  f.Seek(0, 0)

  // upload the file
  err = uploadFile(f)
  if err != nil {
    return
  }

  return nil
}

multireader.go

func handleUpload(u io.Reader) (err error) {
  // read in the first two bytes
  b := make([]byte, 2)
  _, err = u.Read(b)
  if err != nil {
    return
  }

  // check that they match the JPEG header
  jpg := []byte{0xFF, 0xD8}
  if !bytes.Equal(b, jpg) {
    return errors.New("not a JPEG")
  }

  // glue those bytes back onto the reader
  r := io.MultiReader(bytes.NewReader(b), u)

  // upload the file
  err = uploadFile(r)
  if err != nil {
    return
  }

  return nil
}

multiwriter.go

func handleUpload(u io.Reader) (err error) {
  // create the pipes
  mp4R, mp4W := io.Pipe()
  webmR, webmW := io.Pipe()
  oggR, oggW := io.Pipe()
  wavR, wavW := io.Pipe()

  // create channels to synchronize
  done := make(chan bool)
  errs := make(chan error)
  defer close(done)
  defer close(errs)

  // spawn all the task goroutines. These look identical to
  // the TeeReader example, but pulled out into separate
  // methods for clarity
  go uploadMP4(mp4R, done, errs)
  go transcodeAndUploadWebM(webmR, done, errs)
  go transcodeAndUploadOgg(oggR, done, errs)
  go transcodeAndUploadWav(wavR, done, errs)

  go func() {
    // after completing the copy, we need to close
    // the PipeWriters to propagate the EOF to all
    // PipeReaders to avoid deadlock
    defer mp4W.Close()
    defer webmW.Close()
    defer oggW.Close()
    defer wavW.Close()

    // build the multiwriter for all the pipes
    mw := io.MultiWriter(mp4W, webmW, oggW, wavW)

    // copy the data into the multiwriter
    _, err := io.Copy(mw, u)
    if err != nil {
      errs <- err
    }
  }()

  // wait until all are done
  // or an error occurs
  for c := 0; c < 4; c++ {
    select {
    case err := <-errs:
      return err
    case <-done:
    }
  }

  return nil
}

teereader.go

func handleUpload(u io.Reader) (err error) {
  // create the pipe and tee reader
  pr, pw := io.Pipe()
  tr := io.TeeReader(u, pw)

  // create channels to synchronize
  done := make(chan bool)
  errs := make(chan error)
  defer close(done)
  defer close(errs)

  go func() {
    // close the PipeWriter after the
    // TeeReader completes to trigger EOF
    defer pw.Close()

    // upload the original MP4 data
    err := uploadFile(tr)
    if err != nil {
      errs <- err
      return
    }

    done <- true
  }()

  go func() {
    // transcode to WebM
    webmr, err := transcode(pr)
    if err != nil {
      errs <- err
      return
    }

    // upload to storage
    err = uploadFile(webmr)
    if err != nil {
      errs <- err
      return
    }

    done <- true
  }()

  // wait until both are done
  // or an error occurs
  for c := 0; c < 2; {
    select {
    case err := <-errs:
      return err
    case <-done:
      c++
    }
  }

  return nil
}