func (s *Subscription) Unsubscribe() error {
	s.mu.Lock()
	if s.status == SUBCLOSED {
		s.mu.Unlock()
		return ErrSubscriptionClosed
	}
	s.mu.Unlock()
	s.triggerOnUnsubscribe(false, false)
	s.centrifuge.unsubscribe(s.channel, func(result UnsubscribeResult, err error) {})
	return nil
}
Unsubscribe allows to unsubscribe from channel.
package centrifuge
import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/centrifugal/protocol"
)
type SubscribeSuccessEvent struct {
	Resubscribed bool
	Recovered    bool
}
type SubscribeErrorEvent struct {
	Error string
}
type UnsubscribeEvent struct{}
type LeaveEvent struct {
	ClientInfo
}
type JoinEvent struct {
	ClientInfo
}
type PublishEvent struct {
	Publication
}
type PublishHandler interface {
	OnPublish(*Subscription, PublishEvent)
}
type JoinHandler interface {
	OnJoin(*Subscription, JoinEvent)
}
type LeaveHandler interface {
	OnLeave(*Subscription, LeaveEvent)
}
type UnsubscribeHandler interface {
	OnUnsubscribe(*Subscription, UnsubscribeEvent)
}
type SubscribeSuccessHandler interface {
	OnSubscribeSuccess(*Subscription, SubscribeSuccessEvent)
}
type SubscribeErrorHandler interface {
	OnSubscribeError(*Subscription, SubscribeErrorEvent)
}
type subscriptionEventHub struct {
	onPublish          PublishHandler
	onJoin             JoinHandler
	onLeave            LeaveHandler
	onUnsubscribe      UnsubscribeHandler
	onSubscribeSuccess SubscribeSuccessHandler
	onSubscribeError   SubscribeErrorHandler
}
func newSubscriptionEventHub() *subscriptionEventHub {
	return &subscriptionEventHub{}
}
func (s *Subscription) OnPublish(handler PublishHandler) {
	s.events.onPublish = handler
}
func (s *Subscription) OnJoin(handler JoinHandler) {
	s.events.onJoin = handler
}
func (s *Subscription) OnLeave(handler LeaveHandler) {
	s.events.onLeave = handler
}
func (s *Subscription) OnUnsubscribe(handler UnsubscribeHandler) {
	s.events.onUnsubscribe = handler
}
func (s *Subscription) OnSubscribeSuccess(handler SubscribeSuccessHandler) {
	s.events.onSubscribeSuccess = handler
}
func (s *Subscription) OnSubscribeError(handler SubscribeErrorHandler) {
	s.events.onSubscribeError = handler
}
const (
	UNSUBSCRIBED = iota
	SUBSCRIBING
	SUBSCRIBED
	SUBERROR
	SUBCLOSED
)
type Subscription struct {
	futureID        uint64
	mu              sync.Mutex
	channel         string
	centrifuge      *Client
	subCloseCh      chan struct{}
	status          int
	events          *subscriptionEventHub
	lastOffset      uint64
	lastEpoch       string
	recover         bool
	err             error
	needResubscribe bool
	subFutures      map[uint64]subFuture
}
type subFuture struct {
	fn      func(error)
	closeCh chan struct{}
}
func newSubFuture(fn func(error)) subFuture {
	return subFuture{fn: fn, closeCh: make(chan struct{})}
}
func (c *Client) newSubscription(channel string) *Subscription {
	s := &Subscription{
		centrifuge:      c,
		channel:         channel,
		subCloseCh:      make(chan struct{}),
		events:          newSubscriptionEventHub(),
		subFutures:      make(map[uint64]subFuture),
		needResubscribe: false,
	}
	return s
}
func (s *Subscription) Channel() string {
	return s.channel
}
func (s *Subscription) nextFutureID() uint64 {
	return atomic.AddUint64(&s.futureID, 1)
}
func (s *Subscription) resolveSubFutures(err error) {
	for _, fut := range s.subFutures {
		fut.fn(err)
		close(fut.closeCh)
	}
	s.subFutures = make(map[uint64]subFuture)
}
func (s *Subscription) Publish(data []byte) (PublishResult, error) {
	s.mu.Lock()
	if s.status == SUBCLOSED {
		s.mu.Unlock()
		return PublishResult{}, ErrSubscriptionClosed
	}
	s.mu.Unlock()
	resCh := make(chan PublishResult, 1)
	errCh := make(chan error, 1)
	s.publish(data, func(result PublishResult, err error) {
		resCh <- result
		errCh <- err
	})
	return <-resCh, <-errCh
}
type HistoryOptions struct {
	Limit   int32
	Since   *StreamPosition
	Reverse bool
}
type HistoryOption func(options *HistoryOptions)
func WithHistorySince(sp *StreamPosition) HistoryOption {
	return func(options *HistoryOptions) {
		options.Since = sp
	}
}
func WithHistoryLimit(limit int32) HistoryOption {
	return func(options *HistoryOptions) {
		options.Limit = limit
	}
}
func WithHistoryReverse(reverse bool) HistoryOption {
	return func(options *HistoryOptions) {
		options.Reverse = reverse
	}
}
func (s *Subscription) History(opts ...HistoryOption) (HistoryResult, error) {
	historyOpts := &HistoryOptions{}
	for _, opt := range opts {
		opt(historyOpts)
	}
	s.mu.Lock()
	if s.status == SUBCLOSED {
		s.mu.Unlock()
		return HistoryResult{}, ErrSubscriptionClosed
	}
	s.mu.Unlock()
	resCh := make(chan HistoryResult, 1)
	errCh := make(chan error, 1)
	s.history(*historyOpts, func(result HistoryResult, err error) {
		resCh <- result
		errCh <- err
	})
	return <-resCh, <-errCh
}
func (s *Subscription) Presence() (PresenceResult, error) {
	s.mu.Lock()
	if s.status == SUBCLOSED {
		s.mu.Unlock()
		return PresenceResult{}, ErrSubscriptionClosed
	}
	s.mu.Unlock()
	resCh := make(chan PresenceResult, 1)
	errCh := make(chan error, 1)
	s.presence(func(result PresenceResult, err error) {
		resCh <- result
		errCh <- err
	})
	return <-resCh, <-errCh
}
func (s *Subscription) PresenceStats() (PresenceStatsResult, error) {
	resCh := make(chan PresenceStatsResult, 1)
	errCh := make(chan error, 1)
	s.presenceStats(func(result PresenceStatsResult, err error) {
		resCh <- result
		errCh <- err
	})
	return <-resCh, <-errCh
}
func (s *Subscription) onSubscribe(fn func(err error)) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.status == SUBSCRIBED {
		go fn(nil)
	} else if s.status == SUBERROR {
		go fn(s.err)
	} else {
		id := s.nextFutureID()
		fut := newSubFuture(fn)
		s.subFutures[id] = fut
		go func() {
			select {
			case <-fut.closeCh:
			case <-time.After(s.centrifuge.config.ReadTimeout):
				s.mu.Lock()
				defer s.mu.Unlock()
				fut, ok := s.subFutures[id]
				if !ok {
					return
				}
				delete(s.subFutures, id)
				fut.fn(ErrTimeout)
			}
		}()
	}
}
func (s *Subscription) publish(data []byte, fn func(PublishResult, error)) {
	s.onSubscribe(func(err error) {
		if err != nil {
			fn(PublishResult{}, err)
			return
		}
		s.centrifuge.publish(s.channel, data, fn)
	})
}
func (s *Subscription) history(opts HistoryOptions, fn func(HistoryResult, error)) {
	s.onSubscribe(func(err error) {
		if err != nil {
			fn(HistoryResult{}, err)
			return
		}
		s.centrifuge.history(s.channel, opts, fn)
	})
}
func (s *Subscription) presence(fn func(PresenceResult, error)) {
	s.onSubscribe(func(err error) {
		if err != nil {
			fn(PresenceResult{}, err)
			return
		}
		s.centrifuge.presence(s.channel, fn)
	})
}
func (s *Subscription) presenceStats(fn func(PresenceStatsResult, error)) {
	s.onSubscribe(func(err error) {
		if err != nil {
			fn(PresenceStatsResult{}, err)
			return
		}
		s.centrifuge.presenceStats(s.channel, fn)
	})
}