func (s *Stream) Close() error {
	if sid, isOpen := func() (uint16, bool) {
		s.lock.Lock()
		defer s.lock.Unlock()
		isOpen := true
		if s.writeErr == nil {
			s.writeErr = errors.New("Stream closed")
		} else {
			isOpen = false
		}
		if s.readErr == nil {
			s.readErr = io.EOF
		} else {
			isOpen = false
		}
		s.readNotifier.Broadcast()  
		return s.streamIdentifier, isOpen
	}(); isOpen {
		return s.association.sendResetRequest(sid)
	}
	return nil
}
Close closes the write-direction of the stream.
Future calls to Write are not permitted after calling Close.
package sctp
import (
	"io"
	"math"
	"sync"
	"github.com/pion/logging"
	"github.com/pkg/errors"
)
const (
	ReliabilityTypeReliable byte = 0
	ReliabilityTypeRexmit byte = 1
	ReliabilityTypeTimed byte = 2
)
type Stream struct {
	association         *Association
	lock                sync.RWMutex
	streamIdentifier    uint16
	defaultPayloadType  PayloadProtocolIdentifier
	reassemblyQueue     *reassemblyQueue
	sequenceNumber      uint16
	readNotifier        *sync.Cond
	readErr             error
	writeErr            error
	unordered           bool
	reliabilityType     byte
	reliabilityValue    uint32
	bufferedAmount      uint64
	bufferedAmountLow   uint64
	onBufferedAmountLow func()
	log                 logging.LeveledLogger
	name                string
}
func (s *Stream) StreamIdentifier() uint16 {
	s.lock.RLock()
	defer s.lock.RUnlock()
	return s.streamIdentifier
}
func (s *Stream) SetDefaultPayloadType(defaultPayloadType PayloadProtocolIdentifier) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.setDefaultPayloadType(defaultPayloadType)
}
func (s *Stream) setDefaultPayloadType(defaultPayloadType PayloadProtocolIdentifier) {
	s.defaultPayloadType = defaultPayloadType
}
func (s *Stream) SetReliabilityParams(unordered bool, relType byte, relVal uint32) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.setReliabilityParams(unordered, relType, relVal)
}
func (s *Stream) setReliabilityParams(unordered bool, relType byte, relVal uint32) {
	s.unordered = unordered
	s.reliabilityType = relType
	s.reliabilityValue = relVal
}
func (s *Stream) Read(p []byte) (int, error) {
	n, _, err := s.ReadSCTP(p)
	return n, err
}
func (s *Stream) ReadSCTP(p []byte) (int, PayloadProtocolIdentifier, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	for {
		n, ppi, err := s.reassemblyQueue.read(p)
		if err == nil {
			return n, ppi, nil
		} else if err == io.ErrShortBuffer {
			return 0, PayloadProtocolIdentifier(0), err
		}
		err = s.readErr
		if err != nil {
			return 0, PayloadProtocolIdentifier(0), err
		}
		s.readNotifier.Wait()
	}
}
func (s *Stream) handleData(pd *chunkPayloadData) {
	s.lock.Lock()
	defer s.lock.Unlock()
	var readable bool
	if s.reassemblyQueue.push(pd) {
		readable = s.reassemblyQueue.isReadable()
		s.log.Debugf("[%s] reassemblyQueue readable=%v", s.name, readable)
		if readable {
			s.log.Debugf("[%s] readNotifier.signal()", s.name)
			s.readNotifier.Signal()
			s.log.Debugf("[%s] readNotifier.signal() done", s.name)
		}
	}
}
func (s *Stream) handleForwardTSN(newCumulativeTSN uint32, ssn uint16) {
	var readable bool
	s.lock.Lock()
	s.reassemblyQueue.forwardTSN(newCumulativeTSN, s.unordered, ssn)
	readable = s.reassemblyQueue.isReadable()
	s.lock.Unlock()
	if readable {
		s.readNotifier.Signal()
	}
}
func (s *Stream) Write(p []byte) (n int, err error) {
	return s.WriteSCTP(p, s.defaultPayloadType)
}
func (s *Stream) WriteSCTP(p []byte, ppi PayloadProtocolIdentifier) (n int, err error) {
	if len(p) > math.MaxUint16 {
		return 0, errors.Errorf("Outbound packet larger than maximum message size %v", math.MaxUint16)
	}
	s.lock.RLock()
	err = s.writeErr
	s.lock.RUnlock()
	if err != nil {
		return 0, err
	}
	chunks := s.packetize(p, ppi)
	return len(p), s.association.sendPayloadData(chunks)
}
func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPayloadData {
	s.lock.Lock()
	defer s.lock.Unlock()
	i := uint32(0)
	remaining := uint32(len(raw))
	unordered := ppi != PayloadTypeWebRTCDCEP && s.unordered
	var chunks []*chunkPayloadData
	for remaining != 0 {
		fragmentSize := min32(s.association.maxPayloadSize, remaining)
		userData := make([]byte, fragmentSize)
		copy(userData, raw[i:i+fragmentSize])
		chunk := &chunkPayloadData{
			streamIdentifier:     s.streamIdentifier,
			userData:             userData,
			unordered:            unordered,
			beginningFragment:    i == 0,
			endingFragment:       remaining-fragmentSize == 0,
			immediateSack:        false,
			payloadType:          ppi,
			streamSequenceNumber: s.sequenceNumber,
		}
		chunks = append(chunks, chunk)
		remaining -= fragmentSize
		i += fragmentSize
	}
	if !unordered {
		s.sequenceNumber++
	}
	s.bufferedAmount += uint64(len(raw))
	s.log.Tracef("[%s] bufferedAmount = %d", s.name, s.bufferedAmount)
	return chunks
}