func (c *Client) SubscribeChan(stream string, ch chan *Event) error {
	var connected bool
	errch := make(chan error)
	c.mu.Lock()
	c.subscribed[ch] = make(chan bool)
	c.mu.Unlock()

	go func() {
		operation := func() error {
			resp, err := c.request(stream)
			if err != nil {
				c.cleanup(resp, ch)
				return err
			}

			if resp.StatusCode != 200 {
				c.cleanup(resp, ch)
				return errors.New("could not connect to stream")
			}

			if !connected {
				errch <- nil
				connected = true
			}

			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 {
						c.cleanup(resp, ch)
						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)
					}

					select {
					case <-c.subscribed[ch]:
						c.cleanup(resp, ch)
						return nil
					case ch <- msg:
						// message sent
					}
				}
			}
		}

		err := backoff.Retry(operation, backoff.NewExponentialBackOff())
		if err != nil && !connected {
			errch <- err
		}
	}()
	err := <-errch
	close(errch)

	return err
}
func (c *Client) SubscribeChan(stream string, ch chan *Event) error {
	var connected bool
	errch := make(chan error)
	c.mu.Lock()
	c.subscribed[ch] = make(chan bool)
	c.mu.Unlock()

	go func() {
		operation := func() error {
			resp, err := c.request(stream)
			if err != nil {
				c.cleanup(resp, ch)
				return err
			}

			if resp.StatusCode != 200 {
				c.cleanup(resp, ch)
				return errors.New("could not connect to stream")
			}

			if !connected {
				errch <- nil
				connected = true
			}

			reader := NewEventStreamReader(resp.Body)

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

					select {
					case <-c.subscribed[ch]:
						c.cleanup(resp, ch)
						return nil
					case ch <- msg:
						
					}
				}
			}
		}

		err := backoff.Retry(operation, backoff.NewExponentialBackOff())
		if err != nil && !connected {
			errch <- err
		}
	}()
	err := <-errch
	close(errch)

	return err
}
func ( c * Client ) SubscribeChan ( stream string , ch chan * Event ) error { var connected bool errch := make ( chan error ) c . mu . Lock ( ) c . subscribed [ ch ] = make ( chan bool ) c . mu . Unlock ( ) go func ( ) { operation := func ( ) error { resp , err := c . request ( stream ) if err != nil { c . cleanup ( resp , ch ) return err } if resp . StatusCode != 200 { c . cleanup ( resp , ch ) return errors . New ( " " ) } if ! connected { errch <- nil connected = true } 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 { c . cleanup ( resp , ch ) 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 ) } select { case <- c . subscribed [ ch ] : c . cleanup ( resp , ch ) return nil case ch <- msg : // message sent } } } } err := backoff . Retry ( operation , backoff . NewExponentialBackOff ( ) ) if err != nil && ! connected { errch <- err } } ( ) err := <- errch close ( errch ) return err }
// SubscribeChan sends all events to the provided channel
SubscribeChan sends all events to the provided channel