// Read from the socket and subscribe to or unsubscribe from events
196774
func (wsc *wsConnection) readRoutine() {
	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("WSJSONRPC: %v", r)
			}
			wsc.Logger.Error("Panic in WSJSONRPC handler", "err", err, "stack", string(debug.Stack()))
			wsc.WriteRPCResponse(types.RPCInternalError("unknown", err))
			go wsc.readRoutine()
		} else {
			wsc.baseConn.Close() // nolint: errcheck
		}
	}()

	wsc.baseConn.SetPongHandler(func(m string) error {
		return wsc.baseConn.SetReadDeadline(time.Now().Add(wsc.readWait))
	})

	for {
		select {
		case <-wsc.Quit():
			return
		default:
			// reset deadline for every type of message (control or data)
			if err := wsc.baseConn.SetReadDeadline(time.Now().Add(wsc.readWait)); err != nil {
				wsc.Logger.Error("failed to set read deadline", "err", err)
			}
			var in []byte
			_, in, err := wsc.baseConn.ReadMessage()
			if err != nil {
				if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
					wsc.Logger.Info("Client closed the connection")
				} else {
					wsc.Logger.Error("Failed to read request", "err", err)
				}
				wsc.Stop()
				return
			}

			var request types.RPCRequest
			err = json.Unmarshal(in, &request)
			if err != nil {
				wsc.WriteRPCResponse(types.RPCParseError("", errors.Wrap(err, "Error unmarshaling request")))
				continue
			}

			// A Notification is a Request object without an "id" member.
			// The Server MUST NOT reply to a Notification, including those that are within a batch request.
			if request.ID == "" {
				wsc.Logger.Debug("WSJSONRPC received a notification, skipping... (please send a non-empty ID if you want to call a method)")
				continue
			}

			// Now, fetch the RPCFunc and execute it.

			rpcFunc := wsc.funcMap[request.Method]
			if rpcFunc == nil {
				wsc.WriteRPCResponse(types.RPCMethodNotFoundError(request.ID))
				continue
			}
			var args []reflect.Value
			if rpcFunc.ws {
				wsCtx := types.WSRPCContext{Request: request, WSRPCConnection: wsc}
				if len(request.Params) > 0 {
					args, err = jsonParamsToArgsWS(rpcFunc, request.Params, wsCtx)
				}
			} else {
				if len(request.Params) > 0 {
					args, err = jsonParamsToArgsRPC(rpcFunc, request.Params)
				}
			}
			if err != nil {
				wsc.WriteRPCResponse(types.RPCInternalError(request.ID, errors.Wrap(err, "Error converting json params to arguments")))
				continue
			}
			returns := rpcFunc.f.Call(args)

			// TODO: Need to encode args/returns to string if we want to log them
			wsc.Logger.Info("WSJSONRPC", "method", request.Method)

			result, err := unreflectResult(returns)
			if err != nil {
				wsc.WriteRPCResponse(types.RPCInternalError(request.ID, err))
				continue
			} else {
				wsc.WriteRPCResponse(types.NewRPCSuccessResponse(request.ID, result))
				continue
			}

		}
	}
}
390975
func (s *Stream) Read(max int) (events []Event, err error) {
	if max > maxEventBufferSize {
		return nil, ErrMaxBuffer
	}
	if max < minEventBufferSize {
		return nil, ErrMinBuffer
	}
	buffer := make([]C.PmEvent, max)
	numEvents := C.Pm_Read(unsafe.Pointer(s.pmStream), &buffer[0], C.int32_t(max))
	if numEvents < 0 {
		return nil, convertToError(C.PmError(numEvents))
	}
	events = make([]Event, numEvents)
	for i := 0; i < int(numEvents); i++ {
		events[i] = Event{
			Timestamp: Timestamp(buffer[i].timestamp),
			Status:    int64(buffer[i].message) & 0xFF,
			Data1:     (int64(buffer[i].message) >> 8) & 0xFF,
			Data2:     (int64(buffer[i].message) >> 16) & 0xFF,
		}
	}
	return
}
994379
func (c *DefaultClient) UnsubscribeDeviceEvents(appID string, devID string, eventType types.EventType) Token {
	topic := DeviceTopic{appID, devID, DeviceEvents, string(eventType)}
	return c.unsubscribe(topic.String())
}
505639
func (s *Server) Subscribe() (*Client, error) {
	// We'll first atomically obtain the next ID for this client from the
	// incrementing client ID counter.
	clientID := atomic.AddUint64(&s.clientCounter, 1)

	// Create the client that will be returned. The Cancel method is
	// populated to send the cancellation intent to the
	// subscriptionHandler.
	client := &Client{
		updates: queue.NewConcurrentQueue(20),
		quit:    make(chan struct{}),
		Cancel: func() {
			select {
			case s.clientUpdates <- &clientUpdate{
				cancel:   true,
				clientID: clientID,
			}:
			case <-s.quit:
				return
			}
		},
	}

	select {
	case s.clientUpdates <- &clientUpdate{
		cancel:   false,
		clientID: clientID,
		client:   client,
	}:
	case <-s.quit:
		return nil, ErrServerShuttingDown
	}

	return client, nil
}
457055
func (w *Watcher) readEvents() {
	buf := make([]byte, syscall.Getpagesize())

	listener, _ := w.listener.(*netlinkListener)

	for {
		if w.isDone() {
			return
		}

		nr, _, err := syscall.Recvfrom(listener.sock, buf, 0)

		if err != nil {
			w.Error <- err
			continue
		}
		if nr < syscall.NLMSG_HDRLEN {
			w.Error <- syscall.EINVAL
			continue
		}

		msgs, _ := syscall.ParseNetlinkMessage(buf[:nr])

		for _, m := range msgs {
			if m.Header.Type == syscall.NLMSG_DONE {
				w.handleEvent(m.Data)
			}
		}
	}
}
803546
func (client *Client) Listen(socketDefinition protocol.SocketDefinition) (net.Listener, error) {
	// connect to the socket master
	conn, err := net.Dial("tcp", client.socketMasterAddress)
	if err != nil {
		return nil, err
	}

	// bind to a port
	err = protocol.WriteHandshakeRequest(conn, protocol.HandshakeRequest{
		SocketDefinition: socketDefinition,
	})
	if err != nil {
		conn.Close()
		return nil, err
	}

	// see if that worked
	res, err := protocol.ReadHandshakeResponse(conn)
	if err != nil {
		conn.Close()
		return nil, err
	}
	if res.Status != "OK" {
		conn.Close()
		return nil, fmt.Errorf("%s", res.Status)
	}

	// start a new session
	session, err := yamux.Server(conn, yamux.DefaultConfig())
	if err != nil {
		conn.Close()
		return nil, err
	}

	return session, nil
}
265014
func (s *Stream) Listen() <-chan Event {
	ch := make(chan Event)
	go func(s *Stream, ch chan Event) {
		for {
			// sleep for a while before the new polling tick,
			// otherwise operation is too intensive and blocking
			time.Sleep(10 * time.Millisecond)
			events, err := s.Read(1024)
			// Note: It's not very reasonable to push sliced data into
			// a channel, several perf penalities there are.
			// This function is added as a handy utility.
			if err != nil {
				continue
			}
			for i := range events {
				ch <- events[i]
			}
		}
	}(s, ch)
	return ch
}
1351134
func (c *Client) Unsubscribe(listener *Listener) {
	listener.Active = false
	c.tasks <- task{Listener: listener, Action: unsubscribeAction}
}
795105
func (r *Receiver) Listen(h mq.Handler) (retErr error) {
	if !atomic.CompareAndSwapInt32(&r.state, stateCreated, stateConnecting) {
		return errors.New("pubsub: not in created state")
	}

	defer atomic.StoreInt32(&r.state, stateStopped)

	client, err := r.newClient()
	if err != nil {
		return err
	}

	defer func() {
		if err = client.Close(); err != nil && retErr != nil {
			retErr = err
		}
	}()

	atomic.StoreInt32(&r.state, stateListening)
	subscription := client.Subscription(r.subscriptionID)
	defer r.stopRecv()
	err = subscription.Receive(r.recvCtx, func(ctx context.Context, msg *pubsub.Message) {
		wrap := &message{msg: msg}
		h.Handle(wrap)
	})

	if err != context.Canceled {
		return err
	}

	return nil
}
976925
func (lc *LightChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
	return lc.scope.Track(lc.chainHeadFeed.Subscribe(ch))
}
1282583
func (em *EventMeter) Unsubscribe(query string) error {
	em.mtx.Lock()
	defer em.mtx.Unlock()

	return em.wsc.Unsubscribe(context.TODO(), query)
}
1177936
func (c *Client) subscribe() error {
	sub := c.protocol.Subscribe()
	events := sub.Events()

	go func() {
		for {
			select {
			case <-c.quitChan:
				return
			default:
			}
			select {
			case <-c.quitChan:
				return
			case event := <-events:
				switch event.(type) {
				case common.EventNewDevice,
					common.EventNewGroup,
					common.EventNewLocation,
					common.EventExpiredDevice,
					common.EventExpiredGroup,
					common.EventExpiredLocation:
					c.Notify(event)
				}
			}
		}
	}()

	return nil
}
1017727
func (k *KeyLogger) Read() chan InputEvent {
	event := make(chan InputEvent)
	go func(event chan InputEvent) {
		for {
			e, err := k.read()
			if err != nil {
				logrus.Error(err)
				close(event)
				break
			}

			if e != nil {
				event <- *e
			}
		}
	}(event)
	return event
}
716963
func Read(xu *xgbutil.XUtil, block bool) {
	if block {
		ev, err := xu.Conn().WaitForEvent()
		if ev == nil && err == nil {
			xgbutil.Logger.Fatal("BUG: Could not read an event or an error.")
		}
		Enqueue(xu, ev, err)
	}

	// Clean up anything that's in the queue
	for {
		ev, err := xu.Conn().PollForEvent()

		// No events left...
		if ev == nil && err == nil {
			break
		}

		// We're good, queue it up
		Enqueue(xu, ev, err)
	}
}
749310
func (w *Watcher) readEvents() {
	var (
		buf   [syscall.SizeofInotifyEvent * 4096]byte // Buffer for a maximum of 4096 raw events
		n     int                                     // Number of bytes read with read()
		errno error                                   // Syscall errno
	)

	for {
		// See if there is a message on the "done" channel
		select {
		case <-w.done:
			syscall.Close(w.fd)
			close(w.internalEvent)
			close(w.Error)
			return
		default:
		}

		n, errno = syscall.Read(w.fd, buf[:])

		// If EOF is received
		if n == 0 {
			syscall.Close(w.fd)
			close(w.internalEvent)
			close(w.Error)
			return
		}

		if n < 0 {
			w.Error <- os.NewSyscallError("read", errno)
			continue
		}
		if n < syscall.SizeofInotifyEvent {
			w.Error <- errors.New("inotify: short read in readEvents()")
			continue
		}

		var offset uint32 = 0
		// We don't know how many events we just read into the buffer
		// While the offset points to at least one whole event...
		for offset <= uint32(n-syscall.SizeofInotifyEvent) {
			// Point "raw" to the event in the buffer
			raw := (*syscall.InotifyEvent)(unsafe.Pointer(&buf[offset]))
			event := new(FileEvent)
			event.mask = uint32(raw.Mask)
			event.cookie = uint32(raw.Cookie)
			nameLen := uint32(raw.Len)
			// If the event happened to the watched directory or the watched file, the kernel
			// doesn't append the filename to the event, but we would like to always fill the
			// the "Name" field with a valid filename. We retrieve the path of the watch from
			// the "paths" map.
			w.mu.Lock()
			event.Name = w.paths[int(raw.Wd)]
			w.mu.Unlock()
			watchedName := event.Name
			if nameLen > 0 {
				// Point "bytes" at the first byte of the filename
				bytes := (*[syscall.PathMax]byte)(unsafe.Pointer(&buf[offset+syscall.SizeofInotifyEvent]))
				// The filename is padded with NUL bytes. TrimRight() gets rid of those.
				event.Name += "/" + strings.TrimRight(string(bytes[0:nameLen]), "\000")
			}

			// Send the events that are not ignored on the events channel
			if !event.ignoreLinux() {
				// Setup FSNotify flags (inherit from directory watch)
				w.fsnmut.Lock()
				if _, fsnFound := w.fsnFlags[event.Name]; !fsnFound {
					if fsnFlags, watchFound := w.fsnFlags[watchedName]; watchFound {
						w.fsnFlags[event.Name] = fsnFlags
					} else {
						w.fsnFlags[event.Name] = FSN_ALL
					}
				}
				w.fsnmut.Unlock()

				w.internalEvent <- event
			}

			// Move to the next event in the buffer
			offset += syscall.SizeofInotifyEvent + nameLen
		}
	}
}
475247
func (c *Client) listen() {
	c.Server.onNewClientCallback(c)
	reader := bufio.NewReader(c.conn)
	for {
		message, err := reader.ReadString('\n')
		if err != nil {
			c.conn.Close()
			c.Server.onClientConnectionClosed(c, err)
			return
		}
		c.Server.onNewMessage(c, message)
	}
}
694455
func (s *SubSocket) Unsubscribe(topic string) error {
	return s.sock.SetOption(mangos.OptionUnsubscribe, topic)
}