func (p *Port) PullFloat64() (float64, interface{}) {
	item := p.Pull()
	if f, ok := item.(float64); ok {
		return f, nil
	}
	if i, ok := item.(int); ok {
		return float64(i), nil
	}
	return 0, item
}
Pull a float
package core
import (
	"errors"
	"fmt"
	"sync"
	"time"
)
const (
	TYPE_GENERIC   = iota
	TYPE_PRIMITIVE = iota
	TYPE_TRIGGER   = iota
	TYPE_NUMBER    = iota
	TYPE_STRING    = iota
	TYPE_BINARY    = iota
	TYPE_BOOLEAN   = iota
	TYPE_STREAM    = iota
	TYPE_MAP       = iota
)
const (
	_             = iota
	DIRECTION_IN  = iota
	DIRECTION_OUT = iota
)
var CHANNEL_SIZE = 1 << 15
var CHANNEL_DYNAMIC = false
type BOS struct {
	src *Port
}
type EOS struct {
	src *Port
}
type PH struct {
	t string
}
var PHSingle = &PH{"..."}
var PHMultiple = &PH{"[...]"}
type Port struct {
	operator  *Operator
	service   *Service
	delegate  *Delegate
	dests     map[*Port]bool
	src       *Port
	strSrc    *Port
	direction int
	itemType int
	parStr *Port
	parMap *Port
	sub  *Port
	subs map[string]*Port
	buf    chan interface{}
	mutex  sync.Mutex
	closed bool
}
func NewPort(srv *Service, del *Delegate, def TypeDef, dir int) (*Port, error) {
	if !def.valid {
		err := def.Validate()
		if err != nil {
			return nil, err
		}
	}
	if dir != DIRECTION_IN && dir != DIRECTION_OUT {
		return nil, errors.New("wrong direction")
	}
	p := &Port{}
	p.strSrc = p
	p.direction = dir
	if srv != nil {
		p.operator = srv.operator
	} else if del != nil {
		p.operator = del.operator
	}
	p.service = srv
	p.delegate = del
	p.dests = make(map[*Port]bool)
	var err error
	switch def.Type {
	case "map":
		p.itemType = TYPE_MAP
		p.subs = make(map[string]*Port)
		for k, e := range def.Map {
			p.subs[k], err = NewPort(srv, del, *e, dir)
			if err != nil {
				return nil, err
			}
			p.subs[k].parStr = p.parStr
			p.subs[k].parMap = p
		}
	case "stream":
		p.itemType = TYPE_STREAM
		p.sub, err = NewPort(srv, del, *def.Stream, dir)
		if err != nil {
			return nil, err
		}
		setParentStreams(p.sub, p)
	case "primitive":
		p.itemType = TYPE_PRIMITIVE
	case "trigger":
		p.itemType = TYPE_TRIGGER
	case "number":
		p.itemType = TYPE_NUMBER
	case "string":
		p.itemType = TYPE_STRING
	case "binary":
		p.itemType = TYPE_BINARY
	case "boolean":
		p.itemType = TYPE_BOOLEAN
	}
	if p.PrimitiveType() && dir == DIRECTION_IN && p.operator != nil && p.operator.function != nil {
		p.buf = make(chan interface{}, CHANNEL_SIZE)
	}
	return p, nil
}
func (p *Port) Type() int {
	return p.itemType
}
func (p *Port) Direction() int {
	return p.direction
}
func (p *Port) ParentStream() *Port {
	return p.parStr
}
func (p *Port) Operator() *Operator {
	return p.operator
}
func (p *Port) Delegate() *Delegate {
	return p.delegate
}
func (p *Port) Map(name string) *Port {
	port, _ := p.subs[name]
	return port
}
func (p *Port) MapLength() int {
	return len(p.subs)
}
func (p *Port) MapEntryNames() []string {
	entries := []string{}
	for entry := range p.subs {
		entries = append(entries, entry)
	}
	return entries
}
func (p *Port) Stream() *Port {
	return p.sub
}
func (p *Port) Connect(q *Port) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = errors.New(fmt.Sprintf("%s", e))
		}
	}()
	if q.src != nil {
		if q.src == p {
			return nil
		}
		return fmt.Errorf("%s -> %s: already connected", p.String(), q.String())
	}
	if q.itemType == TYPE_PRIMITIVE {
		return p.connect(q, true)
	}
	if q.itemType == TYPE_TRIGGER {
		if p.itemType == TYPE_MAP {
			for _, sub := range p.subs {
				return sub.Connect(q)
			}
			return fmt.Errorf("%s -> %s: trigger connected with empty map", p.Name(), q.Name())
		}
		return p.connect(q, true)
	}
	if p.itemType != TYPE_PRIMITIVE && p.itemType != q.itemType || p.itemType == TYPE_PRIMITIVE && !q.PrimitiveType() {
		return fmt.Errorf("%s -> %s: types don't match - %d != %d", p.Name(), q.Name(), p.itemType, q.itemType)
	}
	if p.PrimitiveType() {
		return p.connect(q, true)
	}
	if p.itemType == TYPE_MAP {
		if len(p.subs) != len(q.subs) {
			return fmt.Errorf("%s -> %s: maps are incompatible - unequal lengths %d and %d", p.Name(), q.Name(), len(p.subs), len(q.subs))
		}
		for k, pe := range p.subs {
			qe, ok := q.subs[k]
			if !ok {
				return fmt.Errorf("%s -> %s: maps are incompatible - %s not present", p.Name(), q.Name(), k)
			}
			err := pe.Connect(qe)
			if err != nil {
				return err
			}
		}
		return nil
	}
	if p.itemType == TYPE_STREAM {
		if q.sub == nil {
			return fmt.Errorf("%s -> %s: streams are incompatible - no sub present", p.Name(), q.Name())
		}
		return p.sub.Connect(q.sub)
	}
	if p.itemType == TYPE_GENERIC {
		return fmt.Errorf("%s -> %s: cannot connect generic type", p.Name(), q.Name())
	}
	return fmt.Errorf("%s -> %s: unknown type", p.Name(), q.Name())
}
func (p *Port) Disconnect(q *Port) error {
	if !p.Connected(q) {
		return errors.New("not connected")
	}
	q.src = nil
	delete(p.dests, q)
	return nil
}
func (p *Port) Connected(q *Port) bool {
	if b, ok := p.dests[q]; ok {
		return b && q.src == p
	}
	return false
}
func (p *Port) Open() {
	if !p.closed {
		return
	}
	p.closed = false
	if p.buf != nil {
		p.buf = make(chan interface{}, CHANNEL_SIZE)
	}
	if p.sub != nil {
		p.sub.Open()
	}
	if len(p.subs) > 0 {
		for _, sub := range p.subs {
			sub.Open()
		}
	}
}
func (p *Port) Close() {
	if p.closed {
		return
	}
	p.closed = true
	if p.buf != nil {
		close(p.buf)
	}
	if p.sub != nil {
		p.sub.Close()
	}
	if len(p.subs) > 0 {
		for _, sub := range p.subs {
			sub.Close()
		}
	}
}
func (p *Port) StreamSource() *Port {
	return p.strSrc
}
func (p *Port) SetStreamSource(srcStr *Port) {
	p.strSrc = srcStr
}
func (p *Port) Merge() bool {
	merged := false
	if p.src != nil {
		for dest := range p.dests {
			p.src.dests[dest] = true
			dest.src = p.src
			dest.strSrc = p.strSrc
			merged = true
		}
		p.src.Disconnect(p)
	} else if len(p.dests) != 0 {
		for dest := range p.dests {
			if dest.itemType != TYPE_TRIGGER {
				continue
			}
			p.strSrc.dests[dest] = true
			dest.strSrc = p.strSrc
			merged = true
		}
	}
	if p.sub != nil {
		merged = p.sub.Merge() || merged
	}
	for _, sub := range p.subs {
		merged = sub.Merge() || merged
	}
	return merged
}
func (p *Port) DirectlyConnected() error {
	if p.direction != DIRECTION_IN {
		return errors.New("can only check in ports")
	}
	if p.PrimitiveType() {
		if p.src == nil {
			return errors.New(p.Name() + " not connected")
		}
		if p.src.direction != DIRECTION_OUT && p.src.operator.name != "" {
			return errors.New(p.Name() + " only out ports can be connected with in ports")
		}
		if p.src.src != nil {
			return errors.New(p.Name() + " has connected source " + p.src.Name() + ": " + p.src.src.Name())
		}
		for dest := range p.src.dests {
			if dest == p {
				return nil
			}
		}
		return errors.New(p.Name() + " not connected back from " + p.src.Name())
	}
	if p.sub != nil {
		if err := p.sub.DirectlyConnected(); err != nil {
			return err
		}
		return nil
	}
	for _, sub := range p.subs {
		if err := sub.DirectlyConnected(); err != nil {
			return err
		}
	}
	return nil
}
func (p *Port) assertChannelSpace() {
	c := cap(p.buf)
	if len(p.buf) > c/2 {
		newChan := make(chan interface{}, 2*c)
		p.mutex.Lock()
		for {
			select {
			case i := <-p.buf:
				newChan <- i
			default:
				goto end
			}
		}
	end:
		p.buf = newChan
		p.mutex.Unlock()
	}
}
func (p *Port) WalkPrimitivePorts(handle func(p *Port)) {
	if p.PrimitiveType() {
		handle(p)
	}
	if p.Stream() != nil {
		p.Stream().WalkPrimitivePorts(handle)
	}
	for _, pname := range p.MapEntryNames() {
		p.Map(pname).WalkPrimitivePorts(handle)
	}
}
func (p *Port) Closed() bool {
	return p.closed
}
func (p *Port) Push(item interface{}) {
	if p.closed {
		return
	}
	if p.buf != nil {
		if CHANNEL_DYNAMIC {
			p.assertChannelSpace()
			p.mutex.Lock()
			p.buf <- item
			p.mutex.Unlock()
		} else {
			p.buf <- item
		}
	}
	for dest := range p.dests {
		if dest.Type() == TYPE_TRIGGER || p.PrimitiveType() {
			dest.Push(item)
		}
	}
	if p.PrimitiveType() {
		return
	}
	if p.itemType == TYPE_MAP {
		m, ok := item.(map[string]interface{})
		if !ok {
			for _, sub := range p.subs {
				sub.Push(item)
			}
			return
		}
		for k, i := range m {
			if sub, ok := p.subs[k]; ok {
				sub.Push(i)
			}
		}
		return
	}
	if p.itemType == TYPE_STREAM {
		items, ok := item.([]interface{})
		if !ok {
			p.sub.Push(item)
			return
		}
		p.PushNoTriggerBOS()
		for _, i := range items {
			p.sub.Push(i)
		}
		p.PushEOS()
	}
}
func (p *Port) PushNoTriggerBOS() {
	p.sub.Push(BOS{p.strSrc})
}
func (p *Port) PushBOS() {
	for dest := range p.dests {
		if dest.Type() == TYPE_TRIGGER {
			dest.Push(nil)
		}
	}
	p.sub.Push(BOS{p.strSrc})
}
func (p *Port) PushEOS() {
	p.sub.Push(EOS{p.strSrc})
}
func (p *Port) Pull() interface{} {
	if p.itemType == TYPE_GENERIC {
		panic("cannot pull from generic")
	}
	if p.buf != nil {
		if CHANNEL_DYNAMIC {
			for {
				p.mutex.Lock()
				select {
				case i := <-p.buf:
					p.mutex.Unlock()
					return i
				default:
					p.mutex.Unlock()
				}
				time.Sleep(1 * time.Millisecond)
			}
		} else {
			return <-p.buf
		}
	}
	if p.PrimitiveType() {
		panic("no buffer")
	}
	if p.itemType == TYPE_MAP {
		var mi interface{}
		itemMap := make(map[string]interface{})
		for k, sub := range p.subs {
			i := sub.Pull()
			if i == PHMultiple {
				mi = PHMultiple
				continue
			}
			if bos, ok := i.(BOS); ok {
				mi = bos
				continue
			}
			if eos, ok := i.(EOS); ok {
				mi = eos
				continue
			}
			itemMap[k] = i
		}
		if mi != nil {
			return mi
		}
		return itemMap
	}
	if p.itemType == TYPE_STREAM {
		i := p.sub.Pull()
		if !p.OwnBOS(i) {
			return i
		}
		items := []interface{}{}
		for {
			i := p.sub.Pull()
			if p.OwnEOS(i) {
				return items
			}
			items = append(items, i)
		}
	}
	panic("unknown type")
}