func (r *Rx) Bytes() []byte {
	return r.p.buffer
}
Bytes returns the byte slice on which the pipe operates.
package pipe
type Rx struct {
	p pipe
	tail uint64
	head uint64
}
func (r *Rx) Init(b []byte) {
	r.p.init(b)
	r.tail = 0xfffffffe * jump
	r.head = r.tail
}
func (r *Rx) Pull() []byte {
	if r.head == r.tail+jump {
		return nil
	}
	header := r.p.readAtomic(r.head)
	if header&slotFree != 0 {
		return nil
	}
	payloadSize := header & slotSizeMask
	newHead := r.head + payloadToSlotSize(payloadSize)
	headWrap := (r.head & revolutionMask) | uint64(len(r.p.buffer))
	if int64(newHead-headWrap) >= 0 {
		if int64(newHead-headWrap) > int64(jump) || newHead&offsetMask != 0 {
			return nil
		}
		if r.tail == r.head {
			r.p.writeAtomic(r.head, slotFree|slotToPayloadSize(newHead-r.head))
			r.tail = newHead
		}
		r.head = newHead
		return r.Pull()
	}
	b := r.p.data(r.head, payloadSize)
	r.head = newHead
	return b
}
func (r *Rx) Flush() {
	if r.head == r.tail {
		return
	}
	r.p.writeAtomic(r.tail, slotFree|slotToPayloadSize(r.head-r.tail))
	r.tail = r.head
}