func (d *dataDestinations) Write(ctx *Context, t *Tuple) error {
	d.rwm.RLock()
	shouldUnlock := true
	defer func() {
		if shouldUnlock {
			d.rwm.RUnlock()
		}
	}()
	for d.paused {
		d.rwm.RUnlock()
		d.rwm.Lock()
		for d.paused {
			d.cond.Wait()
		}
		d.rwm.Unlock()
		d.rwm.RLock()
	}
	if len(d.dsts) == 0 {
		atomic.AddInt64(&d.numDropped, 1)
		if ctx.Flags.DestinationlessTupleLog.Enabled() {
			ctx.droppedTuple(t, d.nodeType, d.nodeName, ETOutput, errors.New("no output destination is connected"))
		}
		return nil
	}
	reportFunc := func(dropped *Tuple) {
		ctx.droppedTuple(t, d.nodeType, d.nodeName, ETOutput, errors.New("the output queue is full"))
	}
	if len(d.dsts) > 1 {
		t.Flags.Set(TFShared)
	}
	var closed []string
	for name, dst := range d.dsts {
		if err := dst.write(ctx, t, reportFunc); err != nil {  
			closed = append(closed, name)
		}
	}
	if closed != nil {
		shouldUnlock = false
		d.rwm.RUnlock()
		d.rwm.Lock()
		defer d.rwm.Unlock()
		for _, n := range closed {
			delete(d.dsts, n)
		}
		if len(d.dsts) == 0 && d.callback != nil {
			go d.callback(ddeDisconnect)
		}
	}
	atomic.AddInt64(&d.numSent, 1)
	return nil
}
Write writes tuples to destinations. It doesn't return any error including
errPipeClosed.
package core
import (
	"errors"
	"fmt"
	"reflect"
	"sync"
	"sync/atomic"
	"gopkg.in/sensorbee/sensorbee.v0/data"
)
func newPipe(inputName string, capacity int) (*pipeReceiver, *pipeSender) {
	p := make(chan *Tuple, capacity)  
	r := &pipeReceiver{
		in: p,
	}
	s := &pipeSender{
		inputName: inputName,
		out:       p,
	}
	r.sender = s
	return r, s
}
type pipeReceiver struct {
	in     <-chan *Tuple
	sender *pipeSender
}
func (r *pipeReceiver) close() {
	go func() {
		r.sender.close()
	}()
}
type QueueDropMode int
const (
	DropNone QueueDropMode = iota
	DropLatest
	DropOldest
)
type pipeSender struct {
	cnt int64
	inputName string
	out       chan *Tuple
	dropMode  QueueDropMode
	rwm sync.RWMutex
	registeredDsts []struct {
		registeredName string
		dst            *dataDestinations
	}
	closed bool
}
func (s *pipeSender) Write(ctx *Context, t *Tuple) error {
	return s.write(ctx, t, func(*Tuple) {})
}
func (s *pipeSender) write(ctx *Context, in *Tuple, droppedTuple func(*Tuple)) error {
	s.rwm.RLock()
	defer s.rwm.RUnlock()
	if s.closed {
		return errPipeClosed
	}
	t := in
	if t.Flags.IsSet(TFShared) {
		t = in.ShallowCopy()
	}
	t.InputName = s.inputName
	if s.dropMode == DropNone {
		s.out <- t
	} else {
	sendLoop:
		for {
			select {
			case s.out <- t:
				break sendLoop
			default:
				if s.dropMode == DropLatest {
					droppedTuple(t)
					return nil
				}
				select {
				case dropped := <-s.out:
					droppedTuple(dropped)
				default:  
				}
			}
		}
	}
	atomic.AddInt64(&s.cnt, 1)
	return nil
}
func (s *pipeSender) Close(ctx *Context) error {
	s.close()
	return nil
}
func (s *pipeSender) close() {
	s.rwm.Lock()
	defer s.rwm.Unlock()
	if s.closed {
		return
	}
	s.closed = true
	close(s.out)
	for _, d := range s.registeredDsts {
		go d.dst.remove(d.registeredName)
	}
	s.registeredDsts = nil
}
func (s *pipeSender) registered(name string, dst *dataDestinations) {
	s.rwm.Lock()
	defer s.rwm.Unlock()
	s.registeredDsts = append(s.registeredDsts, struct {
		registeredName string
		dst            *dataDestinations
	}{name, dst})
}
func (s *pipeSender) count() int64 {
	return atomic.LoadInt64(&s.cnt)
}
func (s *pipeSender) queueStatus() (int, int) {
	s.rwm.RLock()
	defer s.rwm.RUnlock()
	if s.closed {
		return 0, 0
	}
	return len(s.out), cap(s.out)
}
func (s *pipeSender) isClosed() bool {
	s.rwm.RLock()
	defer s.rwm.RUnlock()
	return s.closed
}
type dataSources struct {
	numReceived int64
	numErrors   int64
	nodeType NodeType
	nodeName string
	m     sync.RWMutex
	state *topologyStateHolder
	recvs map[string]*pipeReceiver
	msgChs []chan<- *dataSourcesMessage
}
func newDataSources(nodeType NodeType, nodeName string) *dataSources {
	s := &dataSources{
		nodeType: nodeType,
		nodeName: nodeName,
		recvs:    map[string]*pipeReceiver{},
	}
	s.state = newTopologyStateHolder(&s.m)
	return s
}
type dataSourcesMessage struct {
	cmd dataSourcesCommand
	v   interface{}
}
type dataSourcesCommand int
const (
	ddscAddReceiver dataSourcesCommand = iota
	ddscStop
	ddscToggleGracefulStop
	ddscStopOnDisconnect
)
func (s *dataSources) add(name string, r *pipeReceiver) error {
	s.m.Lock()
	defer s.m.Unlock()
	if s.state.getWithoutLock() >= TSStopping {
		return fmt.Errorf("node '%v' already closed its input", s.nodeName)
	}
	if _, ok := s.recvs[name]; ok {
		return fmt.Errorf("node '%v' is already receiving tuples from '%v'", s.nodeName, name)
	}
	s.recvs[name] = r
	if len(s.msgChs) > 0 {
		s.sendMessageWithoutLock(&dataSourcesMessage{
			cmd: ddscAddReceiver,
			v:   r,
		})
	}
	return nil
}
func (s *dataSources) sendMessage(msg *dataSourcesMessage) {
	s.m.Lock()
	defer s.m.Unlock()
	s.sendMessageWithoutLock(msg)
}
func (s *dataSources) sendMessageWithoutLock(msg *dataSourcesMessage) {
	s.state.waitWithoutLock(TSRunning)
	for _, ch := range s.msgChs {
		ch <- msg
	}
}
func (s *dataSources) remove(name string) {
	s.m.Lock()
	defer s.m.Unlock()
	if s.state.getWithoutLock() >= TSStopping {
		return
	}
	r, ok := s.recvs[name]
	if !ok {
		return
	}
	delete(s.recvs, name)
	r.close()  
}
func (s *dataSources) pour(ctx *Context, w Writer, parallelism int) error {
	if parallelism == 0 {
		parallelism = 1
	}
	var (
		wg            sync.WaitGroup
		logOnce       sync.Once
		collectInputs sync.Once
		inputs        []reflect.SelectCase
		threadErr     error
	)
	err := func() error {
		s.m.Lock()
		defer s.m.Unlock()
		if st, err := s.state.checkAndPrepareForRunningWithoutLock(); err != nil {
			switch st {
			case TSRunning, TSPaused:
				return fmt.Errorf("'%v' already started to receive tuples", s.nodeName)
			case TSStopped:
				return fmt.Errorf("'%v' already stopped receiving tuples", s.nodeName)
			default:
				return fmt.Errorf("'%v' has invalid state: %v", s.nodeName, st)
			}
		}
		genCases := func(msgCh <-chan *dataSourcesMessage) []reflect.SelectCase {
			cs := make([]reflect.SelectCase, 0, len(s.recvs)+2)
			cs = append(cs, reflect.SelectCase{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(msgCh),
			})
			cs = append(cs, reflect.SelectCase{
				Dir: reflect.SelectRecv,
			})
			for _, r := range s.recvs {
				cs = append(cs, reflect.SelectCase{
					Dir:  reflect.SelectRecv,
					Chan: reflect.ValueOf(r.in),
				})
			}
			return cs
		}
		var ensureLocked sync.WaitGroup
		for i := 0; i < parallelism; i++ {
			msgCh := make(chan *dataSourcesMessage)
			s.msgChs = append(s.msgChs, msgCh)
			wg.Add(1)
			ensureLocked.Add(1)
			go func() {
				defer wg.Done()
				needDone := true
				defer func() {
					if needDone {
						ensureLocked.Done()
					}
				}()
				cs := genCases(msgCh)
				ensureLocked.Done()
				needDone = false
				ins, err := s.pouringThread(ctx, w, cs)
				collectInputs.Do(func() {
					inputs = ins
				})
				if err != nil {
					logOnce.Do(func() {
						threadErr = err  
						ctx.ErrLog(err).WithFields(nodeLogFields(s.nodeType, s.nodeName)).
							Error("the node stopped with a fatal error")
					})
				}
			}()
		}
		ensureLocked.Wait()
		return nil
	}()
	if err != nil {
		return err
	}
	s.state.Set(TSRunning)
	wg.Wait()
	s.m.Lock()
	defer func() {
		s.state.setWithoutLock(TSStopped)
		s.m.Unlock()
	}()
	drainTargets := make([]reflect.SelectCase, 0, len(s.recvs)+len(inputs))
	drainTargets = append(drainTargets, inputs...)
	for _, r := range s.recvs {
		drainTargets = append(drainTargets, reflect.SelectCase{
			Dir:  reflect.SelectRecv,
			Chan: reflect.ValueOf(r.in),
		})
		r.close()
	}
	s.recvs = nil
	go func() {
		for len(drainTargets) != 0 {
			i, _, ok := reflect.Select(drainTargets)
			if ok {
				continue
			}
			drainTargets[i] = drainTargets[len(drainTargets)-1]
			drainTargets = drainTargets[:len(drainTargets)-1]
		}
	}()
	for _, ch := range s.msgChs {
		close(ch)
	}
	s.msgChs = nil
	return threadErr
}
func (s *dataSources) pouringThread(ctx *Context, w Writer, cs []reflect.SelectCase) (inputs []reflect.SelectCase, retErr error) {
	const (
		message = iota
		defaultCase
		maxControlIndex = defaultCase
	)
	defer func() {
		if e := recover(); e != nil {
			if err, ok := e.(error); ok {
				if !IsFatalError(err) {
					err = FatalError(err)
				}
				retErr = err
			} else {
				retErr = fmt.Errorf("'%v' got an unknown error through panic: %v", s.nodeName, e)
			}
		}
		if len(cs) > maxControlIndex+1 {
			inputs = cs[maxControlIndex+1:]
		}
		go func() {
			cs = cs[:1]  
			for len(cs) != 0 {
				i, _, ok := reflect.Select(cs)
				if ok {
					continue
				}
				cs[i] = cs[len(cs)-1]
				cs = cs[:len(cs)-1]
			}
		}()
	}()
	gracefulStopEnabled := false
	stopOnDisconnect := false
	reportDT := func(t *Tuple, err error) {
		ctx.droppedTuple(t, s.nodeType, s.nodeName, ETInput, err)
	}
receiveLoop:
	for {
		if stopOnDisconnect && len(cs) == maxControlIndex+1 {
			break
		}
		i, v, ok := reflect.Select(cs)  
		if !ok && i != defaultCase {
			if i <= maxControlIndex {
				retErr = FatalError(fmt.Errorf("a controlling channel (%v) of '%v' has been closed", i, s.nodeName))
				return
			}
			cs[i], cs[len(cs)-1] = cs[len(cs)-1], cs[i]
			cs = cs[:len(cs)-1]
			continue
		}
		switch i {
		case message:
			msg, ok := v.Interface().(*dataSourcesMessage)
			if !ok {
				ctx.Log().WithFields(nodeLogFields(s.nodeType, s.nodeName)).
					Warnf("Received an invalid control message in dataSources: %v", v.Interface())
				continue
			}
			switch msg.cmd {
			case ddscAddReceiver:
				c, ok := msg.v.(*pipeReceiver)
				if !ok {
					ctx.Log().WithFields(nodeLogFields(s.nodeType, s.nodeName)).
						Warn("Cannot add a new receiver due to a type error")
					break
				}
				cs = append(cs, reflect.SelectCase{
					Dir:  reflect.SelectRecv,
					Chan: reflect.ValueOf(c.in),
				})
			case ddscStop:
				if !gracefulStopEnabled {
					break receiveLoop
				}
				cs[defaultCase].Dir = reflect.SelectDefault  
			case ddscToggleGracefulStop:
				gracefulStopEnabled = true
			case ddscStopOnDisconnect:
				stopOnDisconnect = true
			}
		case defaultCase:
			break receiveLoop
		default:
			atomic.AddInt64(&s.numReceived, 1)
			t, ok := v.Interface().(*Tuple)
			if !ok {
				atomic.AddInt64(&s.numErrors, 1)
				ctx.Log().WithFields(nodeLogFields(s.nodeType, s.nodeName)).
					Error("Cannot receive a tuple from a receiver due to a type error")
				break
			}
			err := w.Write(ctx, t)
			if err == nil {
				break
			}
			atomic.AddInt64(&s.numErrors, 1)
			switch {
			case IsFatalError(err):
				retErr = err
				reportDT(t, err)
				return
			case IsTemporaryError(err):
				reportDT(t, err)  
			default:
				reportDT(t, err)
			}
		}
	}
	return  
}
func (s *dataSources) enableGracefulStop() {
	s.sendMessage(&dataSourcesMessage{
		cmd: ddscToggleGracefulStop,
	})
}
func (s *dataSources) stopOnDisconnect() {
	s.sendMessage(&dataSourcesMessage{
		cmd: ddscStopOnDisconnect,
	})
}
func (s *dataSources) stop(ctx *Context) {
	s.m.Lock()
	defer s.m.Unlock()
	if stopped, err := s.state.checkAndPrepareForStoppingWithoutLock(false); stopped || err != nil {
		return
	}
	for _, r := range s.recvs {
		r.close()
	}
	s.recvs = nil
	s.sendMessageWithoutLock(&dataSourcesMessage{
		cmd: ddscStop,
	})
	s.state.waitWithoutLock(TSStopped)
}
func (s *dataSources) status() data.Map {
	s.m.Lock()
	defer s.m.Unlock()
	st := data.Map{}
	st["num_received_total"] = data.Int(atomic.LoadInt64(&s.numReceived))
	st["num_errors"] = data.Int(atomic.LoadInt64(&s.numErrors))
	m := make(data.Map, len(s.recvs))
	for name, recv := range s.recvs {
		if recv.sender.isClosed() {
			delete(s.recvs, name)
			continue
		}
		l, c := recv.sender.queueStatus()
		m[name] = data.Map{
			"num_received": data.Int(recv.sender.count() - int64(l)),
			"queue_size":   data.Int(c),
			"num_queued":   data.Int(l),
		}
	}
	st["inputs"] = m
	return st
}
type dataDestinations struct {
	numSent    int64
	numDropped int64
	nodeType NodeType
	nodeName string
	rwm      sync.RWMutex
	cond     *sync.Cond
	dsts     map[string]*pipeSender
	paused   bool
	callback func(ddEvent)
}
type ddEvent int
const (
	ddeNewConn ddEvent = iota
	ddeDisconnect
)
func newDataDestinations(nodeType NodeType, nodeName string) *dataDestinations {
	d := &dataDestinations{
		nodeType: nodeType,
		nodeName: nodeName,
		dsts:     map[string]*pipeSender{},
	}
	d.cond = sync.NewCond(&d.rwm)
	return d
}
func (d *dataDestinations) add(name string, s *pipeSender) error {
	d.rwm.Lock()
	defer d.rwm.Unlock()
	if d.dsts == nil {
		return fmt.Errorf("node '%v' already closed its output", d.nodeName)
	}
	if _, ok := d.dsts[name]; ok {
		return fmt.Errorf("node '%v' already has the destination '%v'", d.nodeName, name)
	}
	d.dsts[name] = s
	s.registered(name, d)
	if d.callback != nil {
		d.callback(ddeNewConn)
	}
	return nil
}
func (d *dataDestinations) remove(name string) {
	d.rwm.Lock()
	defer d.rwm.Unlock()
	if d.dsts == nil {
		return
	}
	dst, ok := d.dsts[name]
	if !ok {
		return
	}
	delete(d.dsts, name)
	dst.close()
	if len(d.dsts) == 0 && d.callback != nil {
		go d.callback(ddeDisconnect)
	}
}
func (d *dataDestinations) len() int {
	d.rwm.RLock()
	defer d.rwm.RUnlock()
	return len(d.dsts)
}