gochan

README

Gochan

Gochan turns io.ReadWriter interface into channels. Gochan does this by encapsulating io.ReadWriter interface into two separate channels for reading and writing.

Data

Data read from the reading channel is wrapped in the ChanData struct which also contains a Err field if there is any error during reading. Data can be fed into the writing channel as []byte directly without any wrapping.

Cleanup

When done with gochan, simply close the writing channel through CloseWriteChan() method or by closing the io.ReadWriter interface. Closing the writing channel won't affect the reading channel, but closing the io.ReadWriter interface causes both reading and writing channels to be closed.
Note:

1. If you closed io.ReadWriter interface and the writing channel still has buffered data at the point, these data will be lost. To prevent this, use CloseWriteChan() method. CloseWriteChan() will block until the writing channel has finished writing buffered data or encountered errors. After then, you can safely close the io.ReadWriter interface.
2. The reading channel won't be closed since it blocks at reading the data from the io.ReadWriter interface. So the cleanest way to cleanup both reading and writing channels is using CloseWriteChan() method followed by a Close() method of the underlying io.ReadWriter interface if any.

Examples

Socket:

conn, err := net.Dial("tcp", "www.example.com:80")
if err != nil {
	t.Fatal("failed to connect to www.example.com :", err)
}
log.Println("connected")
// creates a read channel with read size 512 bytes and buffer size 50,
// and a write channel with buffer size 50
rc, wc := gochan.NewChan(conn, 50, 50, 512)

wc <- []byte("GET / HTTP/1.0\r\nhost:www.example.com\r\n\r\n")
// loop until EOF
for cd := range rc {
    // cd contains data and error if any
    // len(cd.Data) returns the actual data read
	log.Println(string(cd.Data), cd.Err)
}

// closing connection will automatically close two channels
conn.Close()

Write-only channel:

f, err := os.OpenFile("./testfile", os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0660)
if err != nil {
	t.Fatal("failed to open ./testfile :", err)
}
// create a write-only channel with buffer size 50
wc := gochan.NewWriteonlyChan(f, 50)

for i := 0; i < 10; i = i + 1 {
	want += "eat my shorts" + strconv.Itoa(i) + "!!!\n"
	wc <- []byte("eat my shorts" + strconv.Itoa(i) + "!!!\n")
}
// CloseWriteChan will blocks until the buffered data in the channel is written into the connection.
gochan.CloseWriteChan(wc)
f.Close()

log.Println(want)

Read-only channel:

f, err := os.Open("./testfile")
if err != nil {
	t.Fatal("failed to open ./testfile :", err)
}
// create a read-only channel with buffer size 50 and read size 4096
rc := gochan.NewReadonlyChan(f, 50, 4096)

defer f.Close()
for cd := range rc {
	s := string(cd.Data)
	log.Println(s, cd.Err)
}

Timed command execution:

cmd := exec.Command("ls", "-R", "/")
stdout, err := cmd.StdoutPipe()
if err != nil {
	t.Fatal(err)
}
if err := cmd.Start(); err != nil {
	t.Fatal(err)
}
rc := gochan.NewReadonlyChan(stdout, 0, 4096)
timeout := time.After(1 * time.Second)
loop:
for {
	select {
	case cd, ok := <-rc:
		if !ok || cd.Err == io.EOF {
			break loop
		}
		log.Print(string(cd.Data))
	case <-timeout:
		t.Log("sleep took too long")
		cmd.Process.Kill()
		break loop
	}
}
gochan.CloseReadChan(rc)

Pipe:

rf, wf, err := os.Pipe()
if err != nil {
	t.Fatal("failed to create pipe:", err)
}
rc := gochan.NewReadonlyChan(rf, 50, 4096)
wc := gochan.NewWriteonlyChan(wf, 50)
done := make(chan bool)

go func() {
	for i := 0; i < 10; i = i + 1 {
		wc <- []byte("eat my shorts" + strconv.Itoa(i) + "!!!\n")
	}
	gochan.CloseWriteChan(wc)
	wf.Close()
	done <- true
}()

go func() {
	for cd := range rc {
		s := string(cd.Data)
		log.Print(s, cd.Err)
	}
	done <- true
}()

_ = <-done
_ = <-done

Modify read size after the creation of channels:

rf, wf, err := os.Pipe()
if err != nil {
	t.Fatal("failed to create pipe:", err)
}
var wantData = []string{
	"monkeys",
	"are",
	"typing",
	"randomly",
}

var wantDataLen = []uint32{
	uint32(len(wantData[0])),
	uint32(len(wantData[1])),
	uint32(len(wantData[2])),
	uint32(len(wantData[3])),
}

for i := 0; i < len(wantData); i = i + 1 {
	b := make([]byte, 4)
	binary.BigEndian.PutUint32(b, wantDataLen[i])
	wf.Write(b)
	wf.Write([]byte(wantData[i]))
}
wf.Close()

headerSize := 4
// -1 read buffer size creates a readSize-modifiable channel
rc := gochan.NewReadonlyChan(rf, -1, uint32(headerSize))
for {
	//after reading the header, rc will waits for ModiyReadSize to tell it the next read size
	cd := <-rc
	if cd.Err == io.EOF {
		break
	} else if cd.Err != nil || len(cd.Data) != headerSize {
		t.Fatal("failed to read header size", cd)
	}

	payloadSize := binary.BigEndian.Uint32(cd.Data)
	// tell rc the next read size should be payloadSize
	if err := gochan.ModiyReadSize(rc, payloadSize); err != nil {
		t.Fatal("failed to modify read size", err)
	}
	// read payload
	cd = <-rc
	if cd.Err != nil {
		t.Fatal("failed to read data", cd)
	}
	log.Printf("payload size: %d, payload: %s\n", payloadSize, string(cd.Data))
	// now tell rc to read headerSize bytes of data again
	if err := gochan.ModiyReadSize(rc, uint32(headerSize)); err != nil {
		t.Fatal("failed to modify read size", err)
	}
}

see gochan_test.go for complete code.

Documentation

Overview

gochan package is a wrapper around io.ReadWriter interface to create channels for communication. gochan does this by encapsulating io.ReadWriter interface into two separate channels for reading and writing.

Data read from the reading channel is wrapped in the ChanData struct which also contains a Err field if there is any error during reading. Data can be feed into the writing channel as a slice of bytes directly without any wrapping.

When done with gochan, simply close the writing channel through CloseWriteChan() method or by closing the io.ReadWriter interface. You should note that closing the writing channel won't affect the reading channel, but closing the io.ReadWriter interface will causes both reading and writing channels to be closed. Note:

1. If you closed io.ReadWriter interface and the writing channel still has buffered data at the point, these data will be lost.To prevent this, use CloseWriteChan() method. CloseWriteChan() will block until the writing channel has finished writing buffered data or encountered errors. After then, you can safely close the io.ReadWriter interface.
2. The reading channel won't be closed since it blocks at reading the data from the io.ReadWriter interface. So the cleanest way to cleanup both reading and writing channels is using CloseWriteChan() method followed by a Close() method of the underlying io.ReadWriter interface if any.

Internally, gochan uses two goroutines for reading and writing. The reading goroutine will exit when io.EOF or io.ErrClosedPipe is encountered. The writing goroutine will exit when there is any error during the writing.

Index

• func CloseReadChan(rc <-chan ChanData) error
• func CloseWriteChan(wc chan<- []byte) error
• func ModiyReadSize(rc <-chan ChanData, newSize uint32) error
• func NewChan(rw io.ReadWriter, rcBufsize int, wcBufsize uint, readSize uint32) (<-chan ChanData, chan<- []byte)
• func NewReadonlyChan(r io.Reader, rcBufsize int, readSize uint32) <-chan ChanData
• func NewWriteonlyChan(w io.Writer, wcBufsize uint) chan<- []byte
• type ChanData

Functions

func CloseReadChan

func CloseReadChan(rc <-chan ChanData) error

CloseReadChan closes the reading channel rc.

func CloseWriteChan

func CloseWriteChan(wc chan<- []byte) error

CloseWriteChan closes the writing channel wc. it blocks until the writing channel finished writing its buffered data or encountered errors.

func ModiyReadSize

func ModiyReadSize(rc <-chan ChanData, newSize uint32) error

ModiyReadSize takes a read channel and a new read size. it modify the read channel's read size of next read operation. it return error if rc is closed or not a read channel from the gochan package.

func NewChan

func NewChan(rw io.ReadWriter, rcBufsize int, wcBufsize uint, readSize uint32) (<-chan ChanData, chan<- []byte)

NewChan creates two separate channels with buffer size rcBufsize and wcBufsize for reading/writing from/to rw. Unbuffered channels will be created if buffer size 0 provided. Especially a negative read channel buffer size will create a unbuffered channel with readSize modification support, e.g.:

		headerSize := 4
     // a negative -1 read buffer size will create a unbuffered channel which only reads data once after creation
     // or ModifyReadSize function being called
     rc, _ := gochan.NewChan(rw, -1, 10, headerSize)
     for {
			cd := <- rc
			if cd.Err != nil || len(cd.Data) != 4 {
	 			// handle errror...
			}
         dataSize := binary.BigEndian.Uint32(cd.Data)
			// now make it read a payload size of data
			gochan.ModifyReadSize(rc, dataSize)
         // read the payload
         cd = <- rc
         // make it read a header size of data again
         gochan.ModifyRedSize(rc, headerSize)
         // do what you want with the payload
		}

The readSize parameter indicates the number of bytes a io.Reader.Read() operation will perform. The reading channel returned is typed by ChanData struct, while the writing one is typed by []byte. Every read operation on the channel will return a ChanData containing data and error if any, the actual bytes read can be obtained from len(ChanData.Data). Every write operation should provided a ChanData containing a slice of bytes and the Err field in the ChanData struct will be ignored.

func NewReadonlyChan

func NewReadonlyChan(r io.Reader, rcBufsize int, readSize uint32) <-chan ChanData

NewReadonlyChan creates a readonly channel typed by ChanData struct. see comments on NewChan() method for details on parameters.

func NewWriteonlyChan

func NewWriteonlyChan(w io.Writer, wcBufsize uint) chan<- []byte

NewWriteonlyChan creates a writeonly channel typed by []byte. see comments on NewChan() method for details on parameters.

Types

type ChanData

type ChanData struct {
	Data []byte
	Err  error
}

ChanData contains data and error(if any) for transmission