// notify the session that a stream has closed
768101
func (s *Session) streamClosed(sid uint32) {
	s.streamLock.Lock()
	if n := s.streams[sid].recycleTokens(); n > 0 { // return remaining tokens to the bucket
		if atomic.AddInt32(&s.bucket, int32(n)) > 0 {
			s.notifyBucket()
		}
	}
	delete(s.streams, sid)
	s.streamLock.Unlock()
}
1227990
func (c *serviceStreams) Close() error {
	req := closeReq{
		c.myCode,
		make(chan bool, 1),
	}
	c.r.closeRequests <- req
	<-req.resp
	return nil
}
970827
func (w *gzipResponseWriter) CloseNotify() <-chan bool {
	if cn, ok := w.ResponseWriter.(http.CloseNotifier); ok {
		return cn.CloseNotify()
	}
	panic(httpserver.NonCloseNotifierError{Underlying: w.ResponseWriter})
}
1004010
func (c *Connector) NotifySessionData(ctx context.Context, data *SessionData) {
	s := pitaya.GetSessionFromCtx(ctx)
	err := s.SetData(data.Data)
	if err != nil {
		fmt.Println("got error on notify", err)
	}
}
235704
func (s *Stream) Close() error {
	closeStream := false
	s.stateLock.Lock()
	switch s.state {
	// Opened means we need to signal a close
	case streamSYNSent:
		fallthrough
	case streamSYNReceived:
		fallthrough
	case streamEstablished:
		s.state = streamLocalClose
		goto SEND_CLOSE

	case streamLocalClose:
	case streamRemoteClose:
		s.state = streamClosed
		closeStream = true
		goto SEND_CLOSE

	case streamClosed:
	case streamReset:
	default:
		panic("unhandled state")
	}
	s.stateLock.Unlock()
	return nil
SEND_CLOSE:
	s.stateLock.Unlock()
	s.sendClose()
	s.notifyWaiting()
	if closeStream {
		s.session.closeStream(s.id)
	}
	return nil
}
1015259
func (rww *ResponseWriterWrapper) CloseNotify() <-chan bool {
	if cn, ok := rww.ResponseWriter.(http.CloseNotifier); ok {
		return cn.CloseNotify()
	}
	panic(NonCloseNotifierError{Underlying: rww.ResponseWriter})
}
999825
func (s *Stream) Close() error {
	err := s.stream.Close()

	s.state.Lock()
	switch s.state.v {
	case streamCloseRead:
		s.state.v = streamCloseBoth
		s.remove()
	case streamOpen:
		s.state.v = streamCloseWrite
	}
	s.state.Unlock()
	return err
}
987628
func (s *Session) Close() error {
	s.shutdownLock.Lock()
	defer s.shutdownLock.Unlock()

	if s.shutdown {
		return nil
	}
	s.shutdown = true
	if s.shutdownErr == nil {
		s.shutdownErr = ErrSessionShutdown
	}
	close(s.shutdownCh)
	s.conn.Close()
	<-s.recvDoneCh

	s.streamLock.Lock()
	defer s.streamLock.Unlock()
	for _, stream := range s.streams {
		stream.forceClose()
	}
	return nil
}
209701
func (l *Live) StreamsNotifyUrlConfig(resp interface{}) (err error) {
	req := l.cloneRequest(DescribeLiveStreamsNotifyUrlConfigAction)
	req.AppName = ""
	err = l.rpc.Query(req, resp)
	return
}
339552
func (s *Streamer) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
	// We need to be able to flush for SSE
	fl, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "Flushing not supported", http.StatusNotImplemented)
		return
	}

	// Returns a channel that blocks until the connection is closed
	cn, ok := w.(http.CloseNotifier)
	if !ok {
		http.Error(w, "Closing not supported", http.StatusNotImplemented)
		return
	}
	close := cn.CloseNotify()

	// Set headers for SSE
	h := w.Header()
	h.Set("Cache-Control", "no-cache")
	h.Set("Connection", "keep-alive")
	h.Set("Content-Type", "text/event-stream")

	// Connect new client
	cl := make(client, s.bufSize)
	s.connecting <- cl

	for {
		select {
		case <-close:
			// Disconnect the client when the connection is closed
			s.disconnecting <- cl
			return

		case event := <-cl:
			// Write events
			w.Write(event) // TODO: error handling
			fl.Flush()
		}
	}
}
1158442
func (s *Session) closeStream(id uint32) {
	s.streamLock.Lock()
	if _, ok := s.inflight[id]; ok {
		select {
		case <-s.synCh:
		default:
			s.logger.Printf("[ERR] yamux: SYN tracking out of sync")
		}
	}
	delete(s.streams, id)
	s.streamLock.Unlock()
}
1255304
func (c *Client) CloseSession() {
	c.Pool.Close()
	currClient = nil
	once = sync.Once{}
}
1208809
func (s *Session) Close() {
	atomic.AddInt64(&SessionCount, -1)
	sessionsByID.Delete(s.ID())
	sessionsByUID.Delete(s.UID())
	// TODO: this logic should be moved to nats rpc server
	if s.IsFrontend && s.Subscriptions != nil && len(s.Subscriptions) > 0 {
		// if the user is bound to an userid and nats rpc server is being used we need to unsubscribe
		for _, sub := range s.Subscriptions {
			err := sub.Unsubscribe()
			if err != nil {
				logger.Log.Errorf("error unsubscribing to user's messages channel: %s, this can cause performance and leak issues", err.Error())
			} else {
				logger.Log.Debugf("successfully unsubscribed to user's %s messages channel", s.UID())
			}
		}
	}
	s.entity.Close()
}
236631
func (t *fileTrafficLog) Closed() {
	bytes, _ := json.MarshalIndent(t.Messages, "", "\t")
	ioutil.WriteFile(t.FileName, bytes, 0644)
}
54199
func streamEventCallback(st C.virStreamPtr, events int, callbackID int) {
	callbackFunc := getCallbackId(callbackID)

	callback, ok := callbackFunc.(StreamEventCallback)
	if !ok {
		panic("Incorrect stream event func callback")
	}

	callback(&Stream{ptr: st}, StreamEventType(events))
}
967104
func (c Client) Notify(serviceMethod string, args interface{}) error {
	req := &rpc.Request{
		ServiceMethod: serviceMethod,
		Seq:           seqNotify,
	}
	return c.codec.WriteRequest(req, args)
}
252337
func (l *Live) DeleteLiveStreamsNotifyUrlConfig(resp interface{}) (err error) {
	req := l.cloneRequest(DeleteLiveStreamsNotifyUrlConfigAction)
	req.AppName = ""
	err = l.rpc.Query(req, resp)
	return
}