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 {
			
			event, err := reader.ReadEvent()
			if err != nil {
				if err == io.EOF {
					return nil
				}

				
				if c.disconnectcb != nil {
					c.disconnectcb(c)
				}

				return err
			}

			
			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
Subscribe to a data stream