func (p *Port) Stream() *Port {
	return p.sub
}
Returns the substream port of this port. Port must be of type stream.
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
}