code: func NewJSONSource(opts ...JSONSourceOption) (*JSONSource, error) {
  	j := &JSONSource{
  		records: make(chan record, 3),
  	}
  	for _, opt := range opts {
  		opt(j)
  	}
  
  	if j.listener == nil {
  		var err error
  		j.listener, err = net.Listen("tcp", j.addr)
  		if err != nil {
  			return nil, err
  		}
  	}
  	j.listener = tcpKeepAliveListener{j.listener.(*net.TCPListener)}
  
  	j.server = &http.Server{
  		Addr:    j.addr,
  		Handler: j,
  	}
  	go func() {
  		err := j.server.Serve(j.listener)
  		if err != nil {
  			j.records <- record{err: errors.Wrap(err, "starting server")}
  			close(j.records)
  		}
  	}()
  	return j, nil
  }
docs: // NewJSONSource creates a JSONSource - it takes JSONSourceOptions which modify
  // its behavior.