(ctx context.Context)
| 48 | } |
| 49 | |
| 50 | func (o *defaultDialer) defaultConn(ctx context.Context) (*URLOpener, error) { |
| 51 | o.mu.Lock() |
| 52 | defer o.mu.Unlock() |
| 53 | |
| 54 | // Re-use the connection if possible. |
| 55 | if o.opener != nil && o.conn != nil && !o.conn.IsClosed() { |
| 56 | return o.opener, nil |
| 57 | } |
| 58 | |
| 59 | // First time through, or last time resulted in an error, or connection |
| 60 | // was closed. Initialize the connection. |
| 61 | serverURL := os.Getenv("RABBIT_SERVER_URL") |
| 62 | if serverURL == "" { |
| 63 | return nil, errors.New("RABBIT_SERVER_URL environment variable not set") |
| 64 | } |
| 65 | conn, err := amqp.Dial(serverURL) |
| 66 | if err != nil { |
| 67 | return nil, fmt.Errorf("failed to dial RABBIT_SERVER_URL %q: %w", serverURL, err) |
| 68 | } |
| 69 | o.conn = conn |
| 70 | o.opener = &URLOpener{Connection: conn} |
| 71 | return o.opener, nil |
| 72 | } |
| 73 | |
| 74 | func (o *defaultDialer) OpenTopicURL(ctx context.Context, u *url.URL) (*pubsub.Topic, error) { |
| 75 | opener, err := o.defaultConn(ctx) |
no test coverage detected