func NewOutputStream(streamChan chan string) *OutputStream {
	out := &OutputStream{
		streamChan: streamChan,
		bufSize:  DEFAULT_LINE_BUFFER_SIZE,
		buf:      make([]byte, DEFAULT_LINE_BUFFER_SIZE),
		lastChar: 0,
	}
	return out
}
NewOutputStream creates a new streaming output on the given channel. The
caller must begin receiving on the channel before the command is started.
The OutputStream never closes the channel.
package overseer
import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"
)
const defaultDelayStart uint = 25
type Cmd struct {
	*sync.Mutex
	stateLock  *sync.Mutex
	Name       string
	Group      string
	Args       []string
	Env        []string
	Dir        string
	DelayStart uint         
	RetryTimes uint         
	Stdout     chan string  
	Stderr     chan string  
	State      CmdState     
	startTime  time.Time
	stdout     *OutputBuffer  
	stderr     *OutputBuffer  
	status     Status
	statusChan chan Status    
	changeChan chan CmdState  
	doneChan   chan struct{}  
	buffered   bool           
}
type Status struct {
	Cmd     string
	PID     int
	Exit    int       
	Error   error     
	StartTs int64     
	StopTs  int64     
	Runtime float64   
	Stdout  []string  
	Stderr  []string  
}
type Options struct {
	Group      string
	Dir        string
	Env        []string
	DelayStart uint
	RetryTimes uint
	Buffered bool
	Streaming bool
}
func NewCmd(name string, args ...interface{}) *Cmd {
	var para []string
	var opts Options
	for _, arg := range args {
		switch arg.(type) {
		case []string:
			for _, v := range arg.([]string) {
				para = append(para, fmt.Sprint(v))
			}
		case Options:
			opts = arg.(Options)
		default:
		}
	}
	c := &Cmd{
		Name:       name,
		Group:      opts.Group,
		Args:       para,
		Dir:        opts.Dir,
		Env:        opts.Env,
		DelayStart: defaultDelayStart,
		RetryTimes: opts.RetryTimes,
		Mutex:      &sync.Mutex{},
		stateLock:  &sync.Mutex{},
		status: Status{
			Cmd:     name,
			PID:     0,
			Exit:    -1,
			Error:   nil,
			Runtime: 0,
		},
		changeChan: make(chan CmdState, 5),
		doneChan:   make(chan struct{}),
	}
	if opts.DelayStart > 0 {
		c.DelayStart = opts.DelayStart
	}
	c.buffered = opts.Buffered
	if opts.Streaming {
		c.Stdout = make(chan string, DEFAULT_STREAM_CHAN_SIZE)
		c.Stderr = make(chan string, DEFAULT_STREAM_CHAN_SIZE)
	}
	return c
}
func (c *Cmd) Clone() *Cmd {
	clone := NewCmd(
		c.Name,
		c.Args,
		Options{
			Group:      c.Group,
			Dir:        c.Dir,
			Env:        c.Env,
			DelayStart: c.DelayStart,
			RetryTimes: c.RetryTimes,
			Buffered:   c.buffered,
			Streaming:  c.Stdout != nil,
		},
	)
	return clone
}
func (c *Cmd) setState(state CmdState) {
	if c.State == state || c.IsFinalState() {
		return
	} else if c.IsInitialState() {
		c.stateLock.Lock()
		c.State = STARTING
		c.stateLock.Unlock()
	} else {
		c.stateLock.Lock()
		c.State = state
		c.stateLock.Unlock()
	}
	c.changeChan <- state
}
func (c *Cmd) IsInitialState() bool {
	c.stateLock.Lock()
	defer c.stateLock.Unlock()
	return c.State == INITIAL
}
func (c *Cmd) IsRunningState() bool {
	c.stateLock.Lock()
	defer c.stateLock.Unlock()
	return c.State == STARTING || c.State == RUNNING
}
func (c *Cmd) IsFinalState() bool {
	c.stateLock.Lock()
	defer c.stateLock.Unlock()
	if c.State == INTERRUPT || c.State == FINISHED || c.State == FATAL {
		return true
	}
	return false
}
func (c *Cmd) getRetryTimes() uint {
	c.Lock()
	defer c.Unlock()
	return c.RetryTimes
}
func (c *Cmd) Start() <-chan Status {
	c.Lock()
	defer c.Unlock()
	if c.statusChan != nil {
		return c.statusChan
	}
	c.statusChan = make(chan Status, 1)
	go c.run()
	return c.statusChan
}
func (c *Cmd) Stop() error {
	c.Lock()
	defer c.Unlock()
	if c.statusChan == nil || c.IsInitialState() || c.IsFinalState() {
		return nil
	}
	c.RetryTimes = 0
	c.setState(STOPPING)
	return sendSignal(-c.status.PID, syscall.SIGTERM)
}
func (c *Cmd) Signal(sig syscall.Signal) error {
	c.Lock()
	defer c.Unlock()
	if c.statusChan == nil || c.IsInitialState() || c.IsFinalState() {
		return nil
	}
	if sig == syscall.SIGTERM {
		c.setState(STOPPING)
	}
	return sendSignal(-c.status.PID, sig)
}
func sendSignal(pid int, sig syscall.Signal) error {
	proc, err := os.FindProcess(pid)
	if err != nil {
		return err
	}
	err = proc.Signal(sig)
	if err != nil {
		return err
	}
	return nil
}
func (c *Cmd) Status() Status {
	c.Lock()
	defer c.Unlock()
	if c.statusChan == nil || c.IsInitialState() {
		return c.status
	}
	if c.IsFinalState() {
		if c.buffered && c.status.Stdout == nil {
			c.status.Stdout = c.stdout.Lines()
			c.status.Stderr = c.stderr.Lines()
			c.stdout = nil  
			c.stderr = nil
		}
	} else {
		c.status.Runtime = time.Since(c.startTime).Seconds()
		if c.buffered {
			if c.stdout != nil {
				c.status.Stdout = c.stdout.Lines()
			}
			if c.stderr != nil {
				c.status.Stderr = c.stderr.Lines()
			}
		}
	}
	return c.status
}
func (c *Cmd) Done() <-chan struct{} {
	return c.doneChan
}
func (c *Cmd) run() {
	defer func() {
		c.statusChan <- c.Status()  
		close(c.doneChan)
		close(c.changeChan)
	}()
	c.Lock()
	c.setState(STARTING)
	c.Unlock()
	cmd := exec.Command(c.Name, c.Args...)
	cmd.SysProcAttr = setSysProcAttr()
	if c.buffered && c.Stdout != nil {
		c.stdout = NewOutputBuffer()
		c.stderr = NewOutputBuffer()
		cmd.Stdout = io.MultiWriter(NewOutputStream(c.Stdout), c.stdout)
		cmd.Stderr = io.MultiWriter(NewOutputStream(c.Stderr), c.stderr)
	} else if c.buffered {
		c.Lock()
		c.stdout = NewOutputBuffer()
		c.stderr = NewOutputBuffer()
		cmd.Stdout = c.stdout
		cmd.Stderr = c.stderr
		c.Unlock()
	} else if c.Stdout != nil {
		cmd.Stdout = NewOutputStream(c.Stdout)
		cmd.Stderr = NewOutputStream(c.Stderr)
	} else {
		cmd.Stdout = nil
		cmd.Stderr = nil
	}
	cmd.Env = c.Env
	cmd.Dir = c.Dir
	now := time.Now()
	if err := cmd.Start(); err != nil {
		c.Lock()
		c.status.Error = err
		c.status.StartTs = now.UnixNano()
		c.status.StopTs = time.Now().UnixNano()
		c.setState(FATAL)
		c.Unlock()
		return
	}
	c.Lock()
	c.startTime = now               
	c.status.PID = cmd.Process.Pid  
	c.status.StartTs = now.UnixNano()
	c.setState(RUNNING)
	c.Unlock()
	err := cmd.Wait()
	now = time.Now()
	exitCode := 0
	c.Lock()
	defer c.Unlock()
	if err != nil && fmt.Sprintf("%T", err) == "*exec.ExitError" {
		exiterr := err.(*exec.ExitError)
		err = nil
		if waitStatus, ok := exiterr.Sys().(syscall.WaitStatus); ok {
			exitCode = waitStatus.ExitStatus()  
			if waitStatus.Signaled() {
				err = errors.New(exiterr.Error())  
				c.setState(INTERRUPT)
			}
		}
	}
	c.status.Runtime = now.Sub(c.startTime).Seconds()
	c.status.StopTs = now.UnixNano()
	c.status.Exit = exitCode
	c.status.Error = err
	c.setState(FINISHED)
}
type OutputBuffer struct {
	buf   *bytes.Buffer
	lines []string
	*sync.Mutex
}
func NewOutputBuffer() *OutputBuffer {
	out := &OutputBuffer{
		buf:   &bytes.Buffer{},
		lines: []string{},
		Mutex: &sync.Mutex{},
	}
	return out
}
func (rw *OutputBuffer) Write(p []byte) (n int, err error) {
	rw.Lock()
	n, err = rw.buf.Write(p)  
	rw.Unlock()
	return  
}
func (rw *OutputBuffer) Lines() []string {
	rw.Lock()
	s := bufio.NewScanner(rw.buf)
	for s.Scan() {
		rw.lines = append(rw.lines, s.Text())
	}
	rw.Unlock()
	return rw.lines
}
const (
	DEFAULT_LINE_BUFFER_SIZE = 16384
	DEFAULT_STREAM_CHAN_SIZE = 1000
)
type ErrLineBufferOverflow struct {
	Line       string  
	BufferSize int     
	BufferFree int     
}
func (e ErrLineBufferOverflow) Error() string {
	return fmt.Sprintf("line does not contain newline and is %d bytes too long to buffer (buffer size: %d)",
		len(e.Line)-e.BufferSize, e.BufferSize)
}
type OutputStream struct {
	streamChan chan string
	bufSize    int
	buf        []byte
	lastChar   int
}