// streamEvents receives events over the SSH connection and forwards them to
// the web client.
187460
func (t *TerminalHandler) streamEvents(ws *websocket.Conn, tc *client.TeleportClient) {
	for {
		select {
		// Send push events that come over the events channel to the web client.
		case event := <-tc.EventsChannel():
			data, err := json.Marshal(event)
			if err != nil {
				t.log.Errorf("Unable to marshal audit event %v: %v.", event.GetType(), err)
				continue
			}

			t.log.Debugf("Sending audit event %v to web client.", event.GetType())

			// UTF-8 encode the error message and then wrap it in a raw envelope.
			encodedPayload, err := t.encoder.String(string(data))
			if err != nil {
				t.log.Debugf("Unable to send audit event %v to web client: %v.", event.GetType(), err)
				continue
			}
			envelope := &Envelope{
				Version: defaults.WebsocketVersion,
				Type:    defaults.WebsocketAudit,
				Payload: encodedPayload,
			}
			envelopeBytes, err := proto.Marshal(envelope)
			if err != nil {
				t.log.Debugf("Unable to send audit event %v to web client: %v.", event.GetType(), err)
				continue
			}

			// Send bytes over the websocket to the web client.
			err = websocket.Message.Send(ws, envelopeBytes)
			if err != nil {
				t.log.Errorf("Unable to send audit event %v to web client: %v.", event.GetType(), err)
				continue
			}
		// Once the terminal stream is over (and the close envelope has been sent),
		// close stop streaming envelopes.
		case <-t.terminalContext.Done():
			return
		}
	}
}
1069660
func (b Backend) ProcessEvents(events ...*auditinternal.Event) bool {
	for _, event := range events {
		if event == nil {
			continue
		}
		attr, err := ev.NewAttributes(event)
		if err != nil {
			audit.HandlePluginError(PluginName, err, event)
			continue
		}
		level, stages := b.policyChecker.LevelAndStages(attr)
		if level == auditinternal.LevelNone {
			continue
		}
		// make shallow copy before modifying to satisfy interface definition
		ev := *event
		e, err := policy.EnforcePolicy(&ev, level, stages)
		if err != nil {
			audit.HandlePluginError(PluginName, err, event)
			continue
		}
		if e == nil {
			continue
		}
		b.delegateBackend.ProcessEvents(e)
	}
	// Returning true regardless of results, since dynamic audit backends
	// can never cause apiserver request to fail.
	return true
}
1219823
func FilterEvents(filter EventFilter, userdata interface{}) {
	context := newEventFilterCallbackContext(filter, userdata)
	C.filterEvents(context.cptr())
}
1332943
func (f *Forwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	atomic.AddInt64(&f.stats.totalEventsForwarded, int64(len(events)))
	for _, e := range events {
		if _, err := f.writeString(f.file, e.String()+"\n"); err != nil {
			return errors.Annotate(err, "cannot write event to string")
		}
	}
	return f.file.Sync()
}
1351726
func (sb *StatsdBuffer) SendEvents(events map[string]event.Event) error {
	for _, e := range events {
		sb.eventChannel <- e
	}
	return nil
}
1140219
func (s *Stream) Write(events []Event) error {
	size := len(events)
	if size > maxEventBufferSize {
		return ErrMaxBuffer
	}
	buffer := make([]C.PmEvent, size)
	for i, evt := range events {
		var event C.PmEvent
		event.timestamp = C.PmTimestamp(evt.Timestamp)
		event.message = C.PmMessage((((evt.Data2 << 16) & 0xFF0000) | ((evt.Data1 << 8) & 0xFF00) | (evt.Status & 0xFF)))
		buffer[i] = event
	}
	return convertToError(C.Pm_Write(unsafe.Pointer(s.pmStream), &buffer[0], C.int32_t(size)))
}
909251
func (s *Simulation) PeerEvents(ctx context.Context, ids []enode.ID, filters ...*PeerEventsFilter) <-chan PeerEvent {
	eventC := make(chan PeerEvent)

	// wait group to make sure all subscriptions to admin peerEvents are established
	// before this function returns.
	var subsWG sync.WaitGroup
	for _, id := range ids {
		s.shutdownWG.Add(1)
		subsWG.Add(1)
		go func(id enode.ID) {
			defer s.shutdownWG.Done()

			events := make(chan *simulations.Event)
			sub := s.Net.Events().Subscribe(events)
			defer sub.Unsubscribe()

			subsWG.Done()

			for {
				select {
				case <-ctx.Done():
					if err := ctx.Err(); err != nil {
						select {
						case eventC <- PeerEvent{NodeID: id, Error: err}:
						case <-s.Done():
						}
					}
					return
				case <-s.Done():
					return
				case e := <-events:
					// ignore control events
					if e.Control {
						continue
					}
					match := len(filters) == 0 // if there are no filters match all events
					for _, f := range filters {
						if f.eventType == simulations.EventTypeConn && e.Conn != nil {
							if *f.connUp != e.Conn.Up {
								continue
							}
							// all connection filter parameters matched, break the loop
							match = true
							break
						}
						if f.eventType == simulations.EventTypeMsg && e.Msg != nil {
							if f.msgReceive != nil && *f.msgReceive != e.Msg.Received {
								continue
							}
							if f.protocol != nil && *f.protocol != e.Msg.Protocol {
								continue
							}
							if f.msgCode != nil && *f.msgCode != e.Msg.Code {
								continue
							}
							// all message filter parameters matched, break the loop
							match = true
							break
						}
					}
					var peerID enode.ID
					switch e.Type {
					case simulations.EventTypeConn:
						peerID = e.Conn.One
						if peerID == id {
							peerID = e.Conn.Other
						}
					case simulations.EventTypeMsg:
						peerID = e.Msg.One
						if peerID == id {
							peerID = e.Msg.Other
						}
					}
					if match {
						select {
						case eventC <- PeerEvent{NodeID: id, PeerID: peerID, Event: e}:
						case <-ctx.Done():
							if err := ctx.Err(); err != nil {
								select {
								case eventC <- PeerEvent{NodeID: id, PeerID: peerID, Error: err}:
								case <-s.Done():
								}
							}
							return
						case <-s.Done():
							return
						}
					}
				case err := <-sub.Err():
					if err != nil {
						select {
						case eventC <- PeerEvent{NodeID: id, Error: err}:
						case <-ctx.Done():
							if err := ctx.Err(); err != nil {
								select {
								case eventC <- PeerEvent{NodeID: id, Error: err}:
								case <-s.Done():
								}
							}
							return
						case <-s.Done():
							return
						}
					}
				}
			}
		}(id)
	}

	// wait all subscriptions
	subsWG.Wait()
	return eventC
}
606484
func (h *eventHub) processEvents() {
	for {
		select {
		case connection := <-h.register:
			h.connections.Register(connection)
		case connection := <-h.unregister:
			h.connections.Unregister(connection)
		case msg := <-h.broadcast:
			h.connections.Broadcast(msg)
		}
	}
}
692254
func (c *GorymanClient) QueryEvents(q string) ([]Event, error) {
	query := &proto.Query{}
	query.String_ = pb.String(q)

	message := &proto.Msg{}
	message.Query = query

	response, err := c.sendRecv(message)
	if err != nil {
		return nil, err
	}

	return ProtocolBuffersToEvents(response.GetEvents()), nil
}
1026436
func NewEventPipeline(streams ...EventStream) EventStream {
	switch len(streams) {
	case 0:
		return nil
	case 1:
		return streams[0]
	default:
	}

	streams[0].Subscribe(streams[1])

	return struct {
		EventWriter
		EventEmitter
	}{
		streams[0],
		NewEventPipeline(streams[1:]...),
	}

}
1204204
func (connector *Forwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	atomic.AddInt64(&connector.stats.pipeline, int64(len(events)))
	defer atomic.AddInt64(&connector.stats.pipeline, -int64(len(events)))
	atomic.AddInt64(&connector.stats.totalEventsForwarded, int64(len(events)))
	// could filter here
	if len(events) == 0 {
		return nil
	}
	return connector.sink.AddEvents(ctx, events)
}
180896
func AsyncEvents(e EventStream) EventStream {
	s := &asyncEventStream{
		e:      e,
		events: make(chan Event, 100),
	}
	go s.start()
	return s
}
431376
func (client *sshClient) Stream(cmd string) error {
	// Establish a single command session
	session, err := client.client.NewSession()
	if err != nil {
		return err
	}
	defer session.Close()

	session.Stdout = os.Stdout
	session.Stderr = os.Stderr

	// Execute the command and return any output
	client.logger.Trace("Streaming command on remote server", "cmd", cmd)
	return session.Run(cmd)
}
1035171
func NewEventStream(patient *models.Patient) *EventStream {
	es := new(EventStream)
	es.Patient = patient
	return es
}
1355360
func newEvents(c *Client, ns string) *events {
	return &events{
		client:    c,
		namespace: ns,
	}
}
1242715
func (evs *EventStreamer) Stream(ctx context.Context) error {
	return evs.bls.Stream(ctx)
}
1029829
func writeEvents(client *goes.Client, streamName string) {
	existingEvents := createTestEvents(10, streamName, serverURL, "FooEvent")

	writer := client.NewStreamWriter(streamName)
	err := writer.Append(nil, existingEvents...)
	if err != nil {
		log.Fatal(err)
	}

	ticker := time.NewTicker(time.Second * 2)
	go func() {
		for _ = range ticker.C {
			num := rand.Intn(5)
			newEvents := createTestEvents(num, streamName, serverURL, "FooEvent")
			writer.Append(nil, newEvents...)
		}
	}()

}