Fragment Description:



Processing stream of data on the flight using the io library.
Duplicating io.Readers, using io.Pipe() and io.TeeReader(), can do the trick.
This scheme can be used during file uploads.


ioTeeReaderPiped

Go Playground

Last update, on 2016, Mon 4 Jan, 18:40:28

/* ... <== see fragment description ... */

package main

import (
    "bytes"
    "fmt"
    "io"
    "log"
    "strings"
    // "time"
)

const streamOfData = `This Is A Data Stream Of Whatever`

func main() {
    streamReader := bytes.NewBufferString(streamOfData)
    lowerIt, upperIt, err := handleProcess(streamReader)
    if err != nil {
        fmt.Println(err)
    }
    fmt.Printf("original is: %q\n=-=\n", streamOfData)
    fmt.Printf("LowerIt is: %q\n", lowerIt)
    fmt.Printf("UpperIt is: %q\n=-=\n", upperIt)
}
func handleProcess(r io.Reader) (lower, upper string, err error) {
    // we'll use goroutines to concurrently read from 'pipeR' and 'teeR'
    // as 'pipeW' is written to.
    pipeR, pipeW := io.Pipe()
    teeR := io.TeeReader(r, pipeW)
    // 2 channels to communicate
    lowerChan := make(chan string)
    upperChan := make(chan string)
    // processing 'teeR'
    go func() {
        // !Important
        // to close the writer or else reading from the other end
        // of the pipe will never finish -> deadlock
        defer pipeW.Close()
        upper := onTheFlightUpper(teeR)
        log.Printf("upper in UPPER is: %s\n", upper)
        upperChan <- upper
        close(upperChan)
    }()
    // processing 'pipeR'
    go func() {
        // defer pipeR.Close(), not necessary - closing pipeW will end it
        lower := onTheFlightLower(pipeR)
        log.Printf("lower in LOWER is: %s\n", lower)
        lowerChan <- lower
        close(lowerChan)
    }()
    upper = <-upperChan
    lower = <-lowerChan
    return lower, upper, nil
}
func onTheFlightLower(r io.Reader) string {
    buf := new(bytes.Buffer)
    _, err := buf.ReadFrom(r)
    if err != nil {
        fmt.Printf("Lower in error: %s\n", err)
    }
    return strings.ToLower(buf.String())
}
func onTheFlightUpper(r io.Reader) string {
    buf := new(bytes.Buffer)
    _, err := buf.ReadFrom(r)
    if err != nil {
        fmt.Printf("Upper in error: %s\n", err)
    }
    return strings.ToUpper(buf.String())
}

/*
original is: "This Is A Data Stream Of Whatever"

LowerIt is: "this is a data stream of whatever"
UpperIt is: "THIS IS A DATA STREAM OF WHATEVER"

2016/01/04 15:28:11 upper in ...UPPER is: THIS IS A DATA STREAM OF WHATEVER
2016/01/04 15:28:11 lower in ...LOWER is: this is a data stream of whatever
*/



Comments