client.go
func (c *Client) Subscribe(stream string, handler func(msg *Event)) error {
	operation := func() error {
		resp, err := c.request(stream)
		if err != nil {
			return err
		}
		defer resp.Body.Close()

		reader := NewEventStreamReader(resp.Body)

		for {
			// Read each new line and process the type of event
			event, err := reader.ReadEvent()
			if err != nil {
				if err == io.EOF {
					return nil
				}

				// run user specified disconnect function
				if c.disconnectcb != nil {
					c.disconnectcb(c)
				}

				return err
			}

			// If we get an error, ignore it.
			if msg, err := c.processEvent(event); err == nil {
				if len(msg.ID) > 0 {
					c.EventID = string(msg.ID)
				} else {
					msg.ID = []byte(c.EventID)
				}

				handler(msg)
			}
		}
	}
	return backoff.Retry(operation, backoff.NewExponentialBackOff())
}
func (c *Client) Subscribe(stream string, handler func(msg *Event)) error {
	operation := func() error {
		resp, err := c.request(stream)
		if err != nil {
			return err
		}
		defer resp.Body.Close()

		reader := NewEventStreamReader(resp.Body)

		for {
			// Read each new line and process the type of event
			event, err := reader.ReadEvent()
			if err != nil {
				if err == io.EOF {
					return nil
				}

				// run user specified disconnect function
				if c.disconnectcb != nil {
					c.disconnectcb(c)
				}

				return err
			}

			// If we get an error, ignore it.
			if msg, err := c.processEvent(event); err == nil {
				if len(msg.ID) > 0 {
					c.EventID = string(msg.ID)
				} else {
					msg.ID = []byte(c.EventID)
				}

				handler(msg)
			}
		}
	}
	return backoff.Retry(operation, backoff.NewExponentialBackOff())
}
// Subscribe to a data stream