// NewStreamWriter returns a new *StreamWriter.
553285
func (c *Client) NewStreamWriter(streamName string) *StreamWriter {
	return &StreamWriter{
		client:     c,
		streamName: streamName,
	}
}
178244
func (c *Client) NewStreamReader(streamName string) *StreamReader {
	return &StreamReader{
		streamName: streamName,
		client:     c,
		version:    -1,
		pageSize:   20,
	}
}
104770
func (file File) NewWriter(out output) (wr *Writer, err error) {
	if file.Channels != 1 {
		err = fmt.Errorf("sorry, only mono currently")
		return
	}

	wr = &Writer{}
	wr.output = out
	wr.sampleBuf = bufio.NewWriter(out)
	wr.options = file

	// write header when close to get correct number of samples
	_, err = wr.Seek(12, os.SEEK_SET)
	if err != nil {
		return
	}

	// fmt.Fprintf(wr, "%s", tokenChunkFmt)
	n, err := wr.output.Write(tokenChunkFmt[:])
	if err != nil {
		return
	}
	wr.bytesWritten += n

	chunkFmt := riffChunkFmt{
		LengthOfHeader: 16,
		AudioFormat:    1,
		NumChannels:    file.Channels,
		SampleRate:     file.SampleRate,
		BytesPerSec:    uint32(file.Channels) * file.SampleRate * uint32(file.SignificantBits) / 8,
		BytesPerBloc:   file.SignificantBits / 8 * file.Channels,
		BitsPerSample:  file.SignificantBits,
	}

	err = binary.Write(wr.output, binary.LittleEndian, chunkFmt)
	if err != nil {
		return
	}
	wr.bytesWritten += 20 //sizeof riffChunkFmt

	n, err = wr.output.Write(tokenData[:])
	if err != nil {
		return
	}
	wr.bytesWritten += n

	// leave space for the data size
	_, err = wr.Seek(4, os.SEEK_CUR)
	if err != nil {
		return
	}

	return
}
1218655
func NewBufferWriter(w io.WriteCloser) *BufferWriter {
	return &BufferWriter{
		W: w,
		H: make(http.Header),
	}
}
356940
func NewWriter(wr io.Writer) *Writer {
	return &Writer{writer: wr, lastElement: defaultLastElement, bf: &bytes.Buffer{}}
}
1368804
func NewStream(cfg API, out io.Writer, bufSize int) *Stream {
	return &Stream{
		cfg:       cfg.(*frozenConfig),
		out:       out,
		buf:       make([]byte, 0, bufSize),
		Error:     nil,
		indention: 0,
	}
}
855434
func newStreamConfig(info logger.Info) (*logStreamConfig, error) {
	logGroupName := info.Config[logGroupKey]
	logStreamName, err := loggerutils.ParseLogTag(info, "{{.FullID}}")
	if err != nil {
		return nil, err
	}
	logCreateGroup := false
	if info.Config[logCreateGroupKey] != "" {
		logCreateGroup, err = strconv.ParseBool(info.Config[logCreateGroupKey])
		if err != nil {
			return nil, err
		}
	}

	logNonBlocking := info.Config["mode"] == "non-blocking"

	forceFlushInterval := defaultForceFlushInterval
	if info.Config[forceFlushIntervalKey] != "" {
		forceFlushIntervalAsInt, err := strconv.Atoi(info.Config[forceFlushIntervalKey])
		if err != nil {
			return nil, err
		}
		forceFlushInterval = time.Duration(forceFlushIntervalAsInt) * time.Second
	}

	maxBufferedEvents := int(defaultMaxBufferedEvents)
	if info.Config[maxBufferedEventsKey] != "" {
		maxBufferedEvents, err = strconv.Atoi(info.Config[maxBufferedEventsKey])
		if err != nil {
			return nil, err
		}
	}

	if info.Config[logStreamKey] != "" {
		logStreamName = info.Config[logStreamKey]
	}

	multilinePattern, err := parseMultilineOptions(info)
	if err != nil {
		return nil, err
	}

	containerStreamConfig := &logStreamConfig{
		logStreamName:      logStreamName,
		logGroupName:       logGroupName,
		logCreateGroup:     logCreateGroup,
		logNonBlocking:     logNonBlocking,
		forceFlushInterval: forceFlushInterval,
		maxBufferedEvents:  maxBufferedEvents,
		multilinePattern:   multilinePattern,
	}

	return containerStreamConfig, nil
}
556606
func NewStreamFramer(out chan<- *StreamFrame,
	heartbeatRate, batchWindow time.Duration, frameSize int) *StreamFramer {

	// Create the heartbeat and flush ticker
	heartbeat := time.NewTicker(heartbeatRate)
	flusher := time.NewTicker(batchWindow)

	return &StreamFramer{
		out:        out,
		frameSize:  frameSize,
		heartbeat:  heartbeat,
		flusher:    flusher,
		f:          new(StreamFrame),
		data:       bytes.NewBuffer(make([]byte, 0, 2*frameSize)),
		shutdownCh: make(chan struct{}),
		exitCh:     make(chan struct{}),
	}
}
11127
func NewSnippetWriter(w io.Writer, c *Context, left, right string) *SnippetWriter {
	sw := &SnippetWriter{
		w:       w,
		context: c,
		left:    left,
		right:   right,
		funcMap: template.FuncMap{},
	}
	for name, namer := range c.Namers {
		sw.funcMap[name] = namer.Name
	}
	return sw
}
734478
func newStream(id uint32, frameSize int, sess *Session) *Stream {
	s := new(Stream)
	s.id = id
	s.chReadEvent = make(chan struct{}, 1)
	s.frameSize = frameSize
	s.sess = sess
	s.die = make(chan struct{})
	return s
}
314348
func NewStreamWatcher(d Decoder) *StreamWatcher {
	sw := &StreamWatcher{
		source: d,
		// It's easy for a consumer to add buffering via an extra
		// goroutine/channel, but impossible for them to remove it,
		// so nonbuffered is better.
		result: make(chan Event),
	}
	go sw.receive()
	return sw
}
346839
func newStream(path string, fn streamFunc) *stream {
	return &stream{
		path: path,
		info: streamFuncs.add(fn),
	}
}
54942
func NewWriter(w io.Writer) *Writer {
	enc := New()
	return &Writer{e: *enc, w: csv.NewWriter(w)}
}
1054277
func NewWriter(rc io.WriteCloser) io.WriteCloser {
	return bufWriteCloser{bufio.NewWriter(rc), rc}
}
239041
func NewEventStreamer(cp *mysql.ConnParams, se *schema.Engine, startPos mysql.Position, timestamp int64, sendEvent sendEventFunc) *EventStreamer {
	evs := &EventStreamer{
		sendEvent: sendEvent,
	}
	evs.bls = NewStreamer(cp, se, nil, startPos, timestamp, evs.transactionToEvent)
	evs.bls.extractPK = true
	return evs
}
21272
func NewWriter(w io.Writer) *Writer {
	if wr, ok := w.(*Writer); ok {
		return wr
	}
	return &Writer{
		w:   w,
		buf: make([]byte, 0, DefaultWriterSize),
	}
}
406824
func NewWriter(w io.WriteCloser, bytesPerSec, burstLimit int) *Writer {
	limiter := NewRate(bytesPerSec, burstLimit)

	return &Writer{
		w:       w,
		ctx:     context.Background(),
		limiter: limiter,
	}
}