// Attempt to write to the outgoing channel, within the
// alloted time frame.
517741
func (ctx *rtmCtx) trywrite(e *Event) error {
	tctx, cancel := context.WithTimeout(ctx, ctx.writeTimeout)
	defer cancel()

	select {
	case <-tctx.Done():
		switch err := tctx.Err(); err {
		case context.DeadlineExceeded:
			return errors.New("write timeout")
		default:
			return err
		}
	case ctx.outbuf <- e:
		return nil
	}

	return errors.New("unreachable")
}
978856
func (c *Conn) Write(b []byte) (int, error) {
	if !c.ok() {
		return 0, syscall.EINVAL
	}
	if !c.isOpen() {
		return 0, &net.OpError{
			Op:   "write",
			Net:  c.LocalAddr().Network(),
			Addr: c.LocalAddr(),
			Err:  errClosing,
		}
	}
	c.deadlineMutex.RLock()
	d := timeToDeadline(c.wdeadline)
	c.deadlineMutex.RUnlock()
	return c.writebuf.WriteTimeout(b, d)
}
103032
func (o *Output) Write(p []byte) (n int, err error) {
	o.Lock()
	defer o.Unlock()
	o.used = true
	var firstErr error
	for _, dst := range o.dests {
		_, err := dst.Write(p)
		if err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return len(p), firstErr
}
995909
func (c *Connection) WriteDeadline(d time.Duration) {
	c.log("Write Deadline", d)
	c.dld.wdld = d
	c.dld.wds = true
}
1319151
func outgoing(c *Client) {
	defer c.workers.Done()
	DEBUG.Println(NET, "outgoing started")

	writer := bufio.NewWriter(c.conn)
	for {
		if debugActive() {
			DEBUG.Println(NET, "outgoing waiting for an outbound message")
		}
		select {
		case <-c.stop:
			DEBUG.Println(NET, "outgoing stopped")
			return
		case pub := <-c.obound:
			msg := pub.p.(*packets.PublishPacket)
			if msg.Qos != 0 && msg.MessageID == 0 {
				msg.MessageID = c.getID(pub.t)
				pub.t.(*PublishToken).messageID = msg.MessageID
			}
			//persist_obound(c.persist, msg)

			if c.options.WriteTimeout > 0 {
				c.conn.SetWriteDeadline(time.Now().Add(c.options.WriteTimeout))
			}

			err := msg.Write(writer)
			if err == nil {
				err = writer.Flush()
			}
			if err != nil {
				ERROR.Println(NET, "outgoing stopped with error")
				c.errors <- err
				msg.Release()
				return
			}

			if c.options.WriteTimeout > 0 {
				// If we successfully wrote, we don't want the timeout to happen during an idle period
				// so we reset it to infinite.
				c.conn.SetWriteDeadline(time.Time{})
			}

			if msg.Qos == 0 {
				pub.t.flowComplete()
			}
			if debugActive() {
				DEBUG.Println(NET, "obound wrote msg, id:", msg.MessageID)
			}
			msg.Release()
			packetsSent += 1
		case msg := <-c.oboundP:
			switch msg.p.(type) {
			case *packets.SubscribePacket:
				msg.p.(*packets.SubscribePacket).MessageID = c.getID(msg.t)
			case *packets.UnsubscribePacket:
				msg.p.(*packets.UnsubscribePacket).MessageID = c.getID(msg.t)
			}
			if debugActive() {
				DEBUG.Println(NET, "obound priority msg to write, type", reflect.TypeOf(msg.p))
			}
			err := msg.p.Write(writer)
			msg.p.Release()
			if err == nil {
				writer.Flush()
			}
			if err != nil {
				ERROR.Println(NET, "outgoing stopped with error")
				c.errors <- err
				return
			}
			switch msg.p.(type) {
			case *packets.DisconnectPacket:
				msg.t.(*DisconnectToken).flowComplete()
				if debugActive() {
					DEBUG.Println(NET, "outbound wrote disconnect, stopping")
				}
				return
			}
			packetsSent += 1
		}
		// Reset ping timer after sending control packet.
		if c.resetPing != nil {
			c.resetPing <- struct{}{}
		}
	}
}
482114
func (lp *lookupPeer) Write(data []byte) (int, error) {
	lp.conn.SetWriteDeadline(time.Now().Add(time.Second))
	return lp.conn.Write(data)
}
765087
func (s *Chan) WriteTo(w io.Writer) {
	// new buffer per message
	// if bottleneck, cycle around a set of buffers
	mw := NewWriter(w)

	// single writer, no need for Mutex
	mw.(*writer).lock = new(nullLocker)
Loop:
	for {
		select {
		case <-s.CloseChan:
			break Loop // told we're done

		case msg, ok := <-s.MsgChan:
			if !ok { // chan closed
				break Loop
			}

			if err := mw.WriteMsg(msg); err != nil {
				if err != io.EOF {
					// unexpected error. tell the client.
					s.ErrChan <- err
				}

				break Loop
			}
		}
	}

	// signal we're done
	s.CloseChan <- true
}
1013290
func (mw *Writer) Write(buf []byte) (int, error) {
	if mw.err != nil {
		return 0, mw.err
	}

	var wrCnt int
	for _, b := range buf {
		zeros, ones := numBits(b)

		// If possible, avoid flushing to maintain high efficiency.
		if ensured := mw.bufCnt < EnsureRawBytes; ensured {
			goto skipEncode
		}
		if huffLen, _ := mw.computeHuffLen(mw.buf0s+zeros, mw.buf1s+ones); huffLen > 0 {
			goto skipEncode
		}

		mw.err = mw.encodeBlock(FinalNil)
		if mw.err != nil {
			break
		}

	skipEncode:
		mw.buf0s += zeros
		mw.buf1s += ones
		mw.buf[mw.bufCnt] = b
		mw.bufCnt++
		wrCnt++
	}

	mw.InputOffset += int64(wrCnt)
	return wrCnt, mw.err
}
625661
func (c *channel) WriteExtended(data []byte, extendedCode uint32) (n int, err error) {
	if c.sentEOF {
		return 0, io.EOF
	}
	// 1 byte message type, 4 bytes remoteId, 4 bytes data length
	opCode := byte(msgChannelData)
	headerLength := uint32(9)
	if extendedCode > 0 {
		headerLength += 4
		opCode = msgChannelExtendedData
	}

	c.writeMu.Lock()
	packet := c.packetPool[extendedCode]
	// We don't remove the buffer from packetPool, so
	// WriteExtended calls from different goroutines will be
	// flagged as errors by the race detector.
	c.writeMu.Unlock()

	for len(data) > 0 {
		space := min(c.maxRemotePayload, len(data))
		if space, err = c.remoteWin.reserve(space); err != nil {
			return n, err
		}
		if want := headerLength + space; uint32(cap(packet)) < want {
			packet = make([]byte, want)
		} else {
			packet = packet[:want]
		}

		todo := data[:space]

		packet[0] = opCode
		binary.BigEndian.PutUint32(packet[1:], c.remoteId)
		if extendedCode > 0 {
			binary.BigEndian.PutUint32(packet[5:], uint32(extendedCode))
		}
		binary.BigEndian.PutUint32(packet[headerLength-4:], uint32(len(todo)))
		copy(packet[headerLength:], todo)
		if err = c.writePacket(packet); err != nil {
			return n, err
		}

		n += len(todo)
		data = data[len(todo):]
	}

	c.writeMu.Lock()
	c.packetPool[extendedCode] = packet
	c.writeMu.Unlock()

	return n, err
}
1159978
func (r *FilteredConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
	select {
	case <-r.closed:
		return 0, errClosed
	default:
	}

	if r.filter != nil {
		r.filter.Outgoing(b, addr)
	}
	return r.source.WriteTo(b, addr)
}
168046
func (c timeoutConn) Write(b []byte) (n int, err error) {
	// Reset deadline
	deadline := time.Now().Add(c.timeout)
	err = c.Conn.SetWriteDeadline(deadline)
	if err != nil {
		return
	}

	return c.Conn.Write(b)
}
131509
func (rs *RetryingSink) Write(event Event) error {
	logger := logrus.WithField("event", event)

retry:
	select {
	case <-rs.closed:
		return ErrSinkClosed
	default:
	}

	if backoff := rs.strategy.Proceed(event); backoff > 0 {
		select {
		case <-time.After(backoff):
			// TODO(stevvooe): This branch holds up the next try. Before, we
			// would simply break to the "retry" label and then possibly wait
			// again. However, this requires all retry strategies to have a
			// large probability of probing the sync for success, rather than
			// just backing off and sending the request.
		case <-rs.closed:
			return ErrSinkClosed
		}
	}

	if err := rs.sink.Write(event); err != nil {
		if err == ErrSinkClosed {
			// terminal!
			return err
		}

		logger := logger.WithError(err) // shadow!!

		if rs.strategy.Failure(event, err) {
			logger.Errorf("retryingsink: dropped event")
			return nil
		}

		logger.Errorf("retryingsink: error writing event, retrying")
		goto retry
	}

	rs.strategy.Success(event)
	return nil
}
888580
func (conn *WebsocketConnection) Write(b []byte) (n int, err error) {
	conn.writeMutex.Lock()
	defer conn.writeMutex.Unlock()

	// allocate a writer
	var writer io.WriteCloser
	writer, err = conn.ws.NextWriter(websocket.BinaryMessage)
	if err != nil {
		return
	}

	// write
	n, err = writer.Write(b)
	if err != nil {
		return
	}

	// close it
	err = writer.Close()
	return
}
445768
func (c ConnWithTimeouts) Write(b []byte) (int, error) {
	if c.writeTimeout == 0 {
		return c.Conn.Write(b)
	}
	if err := c.Conn.SetWriteDeadline(time.Now().Add(c.writeTimeout)); err != nil {
		return 0, err
	}
	return c.Conn.Write(b)
}
503145
func (f FailTemporaryChannelFailure) Error() string {
	if f.Update == nil {
		return f.Code().String()
	}

	return fmt.Sprintf("TemporaryChannelFailure(update=%v)",
		spew.Sdump(f.Update))
}
179965
func (cw *wsReadWriteCloser) Write(p []byte) (n int, err error) {
	w, err := cw.wsConn.NextWriter(websocket.TextMessage)
	if err != nil {
		return 0, err
	}
	if n, err = w.Write(p); err != nil {
		return n, err
	}
	err = w.Close()
	return n, err
}
342931
func (s *DeadlineStream) WriteJSON(v interface{}) error {
	// This uses a real clock rather than trying to use a clock passed
	// in because the websocket will use a real clock to determine
	// whether the deadline has passed anyway.
	deadline := time.Now().Add(s.Timeout)
	if err := s.Conn.SetWriteDeadline(deadline); err != nil {
		return errors.Annotate(err, "setting write deadline")
	}
	return errors.Trace(s.Conn.WriteJSON(v))
}