Try support streaming processing in Source and StreamFunction

[fanweixiao]

In that way, we can write code like this:

// User's http request connection
var client net.Conn

// yomo-source
var source Yomo.Source

// yomo-sfn
var sfn Yomo.StreamFunction

iocopy := func(src, dest net.Conn) {
  defer src.Close()
  defer dest.Close()
  io.Copy(src, dest)
}

// pipe response
go iocopy(client, sfn)

// pipe request
go iocopy(source, client)

[fanweixiao]

Example Pipeline show this:

yomo/example/1-pipeline/source/pipe.go

Lines 43 to 59 in 6f527f9

	func processPipe(r io.Reader, w io.Writer) error {
	buf := make([]byte, 4)
	for {
	_, e := r.Read(buf)
	if e != nil {
	log.Printf("\n--ERR--r.Read(): %v", e)
	return e
	}
	// emit data
	_, e = w.Write(buf)
	if e != nil {
	return e
	}
	log.Printf("Read: %# x", buf)
	time.Sleep(1 * time.Second)
	}
	}

This can be done by io.Copy(w io.Writer, r io.Reader)

[wujunzhuo]

Consider using multiple streams and adding an abstract concept Session to manage them:

[wujunzhuo]

And the SyncHandler could be implemented by

Go channel:

func handler(in <-chan []byte, out chan<- []byte) error {
    for {
        req := <-in
        if req == nil {
            // upstream finished
            return nil
        }
        
        // process upstream data

        for {
            var res []byte

            // send to downstream
            out <- res
        }
    }

    return nil
}

Or io.Reader & io.Writer:

func handler(in io.Reader, out io.Writer) error {
    for {
        req := make([]byte, 512)
        n, err := in.Read(buf) // blocking
        if err != nil {
            return err // err may be io.EOF
        }

        // process upstream data

        for {
            var res []byte
            
            // send to downstream
            out.Write(res)
        }
    }

    return nil
}