func (dc *DuplexConnection) Close() error {
	if !dc.closed.CAS(false, true) {
		return nil
	}
	if dc.keepaliver != nil {
		dc.keepaliver.Stop()
	}
	close(dc.sndQueue)
	dc.cond.L.Lock()
	dc.cond.Broadcast()
	dc.cond.L.Unlock()
	var writeDone chan struct{}
	dc.locker.RLock()
	writeDone = dc.writeDone
	dc.locker.RUnlock()
	if writeDone != nil {
		<-dc.writeDone
	}
	dc.destroyTransport()
	err := dc.GetError()
	if err == nil {
		dc.destroyHandler(errSocketClosed)
	} else {
		dc.destroyHandler(err)
	}
	dc.destroySndQueue()
	dc.destroySndBacklog()
	dc.destroyFragment()
	if dc.destroyReqSche {
		_ = dc.reqSche.Close()
	}
	return err
}
Close close current socket.
package socket
import (
	"context"
	"runtime"
	"sync"
	"time"
	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/pkg/errors"
	"go.uber.org/atomic"
	"github.com/rsocket/rsocket-go/core"
	"github.com/rsocket/rsocket-go/core/framing"
	"github.com/rsocket/rsocket-go/core/transport"
	"github.com/rsocket/rsocket-go/internal/bytesconv"
	"github.com/rsocket/rsocket-go/internal/common"
	"github.com/rsocket/rsocket-go/internal/fragmentation"
	"github.com/rsocket/rsocket-go/internal/map32"
	"github.com/rsocket/rsocket-go/internal/misc"
	"github.com/rsocket/rsocket-go/internal/queue"
	"github.com/rsocket/rsocket-go/lease"
	"github.com/rsocket/rsocket-go/logger"
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx"
	"github.com/rsocket/rsocket-go/rx/flux"
	"github.com/rsocket/rsocket-go/rx/mono"
)
const _outChanSize = 64
const _minRequestSchedulerSize = 1000
var errSocketClosed = errors.New("rsocket: socket closed already")
var errRequestFailed = errors.New("rsocket: send request failed")
var (
	unsupportedRequestStream   = []byte("Request-Stream not implemented.")
	unsupportedRequestResponse = []byte("Request-Response not implemented.")
	unsupportedRequestChannel  = []byte("Request-Channel not implemented.")
)
func mustExecute(sc scheduler.Scheduler, handler func()) {
	if err := sc.Worker().Do(handler); err == nil {
		return
	}
	go handler()
}
type DuplexConnection struct {
	ctx            context.Context
	reqSche        scheduler.Scheduler
	resSche        scheduler.Scheduler
	destroyReqSche bool
	locker         sync.RWMutex
	counter        *core.TrafficCounter
	tp             *transport.Transport
	sndQueue       chan core.WriteableFrame
	sndBacklog     []core.WriteableFrame
	responder      Responder
	messages       map32.Map32  
	sids           StreamID
	mtu            int
	fragments      map32.Map32  
	writeDone      chan struct{}
	keepaliver     *Keepaliver
	cond           sync.Cond
	e              error
	leases         lease.Factory
	closed         *atomic.Bool
	ready          *atomic.Bool
}
func (dc *DuplexConnection) SetError(err error) {
	dc.locker.Lock()
	dc.e = err
	dc.locker.Unlock()
}
func (dc *DuplexConnection) GetError() (err error) {
	dc.locker.RLock()
	err = dc.e
	dc.locker.RUnlock()
	return
}
func (dc *DuplexConnection) nextStreamID() (sid uint32) {
	var firstLap bool
	for {
		sid, firstLap = dc.sids.Next()
		if firstLap {
			return
		}
		_, ok := dc.messages.Load(sid)
		if !ok {
			return
		}
	}
}