Repository: knative/pkg
Pull request #1729 (2020-09-22 00:33:38)

Add a func to return a connection which has been established
`NewDurableSendingConnection` creates a managed connection which is connecting asynchronously. We have to wait the connection to be established before sending message.

Proposed to add a function `NewDurableSendingConnectionGuaranteed` to return a connection which can be used immediately by wrapping a wait.

Also remove `IsEstablished` as we can use `Status` instead. Its downstream usage has been removed.

/assign @vagababov 

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1729#" title="Author self-approved">yanweiguo</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **tcnghia**
You can assign the PR to them by writing `/assign @tcnghia` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[websocket/OWNERS](https://github.com/knative/pkg/blob/master/websocket/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["tcnghia"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[websocket/connection.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1729/pull-knative-pkg-go-coverage/1308202957019287553/artifacts/line-cov.html#file0) | 95.1% | 96.2% | 1.1

codecov[bot]: # [Codecov](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=h1) Report
> Merging [#1729](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=desc) into [master](https://codecov.io/gh/knative/pkg/commit/9128d4b9c1f2bfc33086e8c13f1c307d058c5687?el=desc) will **increase** coverage by `0.04%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative/pkg/pull/1729/graphs/tree.svg?width=650&height=150&src=pr&token=rlqpDnfxbO)](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #1729      +/-   ##
==========================================
+ Coverage   67.62%   67.66%   +0.04%     
==========================================
  Files         226      226              
  Lines        9547     9548       +1     
==========================================
+ Hits         6456     6461       +5     
+ Misses       2719     2717       -2     
+ Partials      372      370       -2     
```


| [Impacted Files](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [network/transports.go](https://codecov.io/gh/knative/pkg/pull/1729/diff?src=pr&el=tree#diff-bmV0d29yay90cmFuc3BvcnRzLmdv) | `92.10% <100.00%> (-0.40%)` | :arrow_down: |
| [websocket/connection.go](https://codecov.io/gh/knative/pkg/pull/1729/diff?src=pr&el=tree#diff-d2Vic29ja2V0L2Nvbm5lY3Rpb24uZ28=) | `94.06% <100.00%> (+1.89%)` | :arrow_up: |
| [test/gcs/mock/mock.go](https://codecov.io/gh/knative/pkg/pull/1729/diff?src=pr&el=tree#diff-dGVzdC9nY3MvbW9jay9tb2NrLmdv) | `91.39% <0.00%> (+1.07%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=footer). Last update [6a12c75...26be939](https://codecov.io/gh/knative/pkg/pull/1729?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-prow-robot: @yanweiguo: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 26be939b47a7af2a8ff1dfbdfb705e72f18decc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative_pkg/1729/pull-knative-pkg-unit-tests/1308202957015093248) | `/test pull-knative-pkg-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative_pkg/1729). [Your PR dashboard](https://gubernator.knative.dev/pr/yanweiguo).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1729#" title="Author self-approved">yanweiguo</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **markusthoemmes**
You can assign the PR to them by writing `/assign @markusthoemmes` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[websocket/OWNERS](https://github.com/knative/pkg/blob/master/websocket/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["markusthoemmes"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[websocket/connection.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1729/pull-knative-pkg-go-coverage/1308292169529823237/artifacts/line-cov.html#file0) | 95.1% | 96.2% | 1.1

knative-prow-robot: @yanweiguo: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 98f46023e5f6906fe39dcf6882ffb2566d938ff5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative_pkg/1729/pull-knative-pkg-unit-tests/1308292169529823234) | `/test pull-knative-pkg-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative_pkg/1729). [Your PR dashboard](https://gubernator.knative.dev/pr/yanweiguo).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/.github/workflows/knative-go-build.yaml b/.github/workflows/knative-go-build.yaml
index 98c9e411a..609551157 100644
--- a/.github/workflows/knative-go-build.yaml
+++ b/.github/workflows/knative-go-build.yaml
@@ -27,7 +27,7 @@ jobs:
     name: Build
     strategy:
       matrix:
-        go-version: [1.14.x]
+        go-version: [1.15.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index 49ef66bf1..31a0aeb2e 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -29,10 +29,10 @@ jobs:
 
     steps:
 
-      - name: Set up Go 1.14.x
+      - name: Set up Go 1.15.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.14.x
+          go-version: 1.15.x
         id: go
 
       - name: Check out code
@@ -83,11 +83,10 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-
-      - name: Set up Go 1.14.x
+      - name: Set up Go 1.15.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.14.x
+          go-version: 1.15.x
         id: go
 
       - name: Check out code
diff --git a/.github/workflows/knative-verify.yaml b/.github/workflows/knative-verify.yaml
index c29a6a04d..57b541280 100644
--- a/.github/workflows/knative-verify.yaml
+++ b/.github/workflows/knative-verify.yaml
@@ -27,7 +27,7 @@ jobs:
     name: Verify Deps and Codegen
     strategy:
       matrix:
-        go-version: [1.14.x]
+        go-version: [1.15.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
diff --git a/.golangci.yaml b/.golangci.yaml
index a33656679..9d60f5792 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -7,7 +7,16 @@ run:
 
 linters:
   enable:
-    - unconvert
+    - gosec
     - prealloc
+    - unconvert
+    - unparam
   disable:
     - errcheck
+
+issues:
+  exclude-rules:
+    - path: test # Excludes /test, *_test.go etc.
+      linters:
+        - gosec
+        - unparam
diff --git a/codegen/cmd/injection-gen/generators/packages.go b/codegen/cmd/injection-gen/generators/packages.go
index ab2ee1205..70e4eb7d1 100644
--- a/codegen/cmd/injection-gen/generators/packages.go
+++ b/codegen/cmd/injection-gen/generators/packages.go
@@ -572,7 +572,7 @@ func reconcilerPackages(basePackage string, groupPkgName string, gv clientgentyp
 	return vers
 }
 
-func versionDuckPackages(basePackage string, groupPkgName string, gv clientgentypes.GroupVersion, groupGoName string, boilerplate []byte, typesToGenerate []*types.Type, customArgs *informergenargs.CustomArgs) []generator.Package {
+func versionDuckPackages(basePackage string, groupPkgName string, gv clientgentypes.GroupVersion, groupGoName string, boilerplate []byte, typesToGenerate []*types.Type, _ *informergenargs.CustomArgs) []generator.Package {
 	packagePath := filepath.Join(basePackage, "ducks", groupPkgName, strings.ToLower(gv.Version.NonEmpty()))
 
 	vers := make([]generator.Package, 0, 2*len(typesToGenerate))
diff --git a/configmap/hash-gen/main.go b/configmap/hash-gen/main.go
index 8d9b7bfe0..bde4f7871 100644
--- a/configmap/hash-gen/main.go
+++ b/configmap/hash-gen/main.go
@@ -49,6 +49,7 @@ func processFile(fileName string) error {
 		return err
 	}
 
+	// nolint:gosec // This is not security critical so open permissions are fine.
 	if err := ioutil.WriteFile(fileName, out, 0644); err != nil {
 		return fmt.Errorf("failed to write file: %w", err)
 	}
diff --git a/kmeta/names.go b/kmeta/names.go
index c59090b52..43d12de80 100644
--- a/kmeta/names.go
+++ b/kmeta/names.go
@@ -17,7 +17,7 @@ limitations under the license.
 package kmeta
 
 import (
-	"crypto/md5"
+	"crypto/md5" // nolint:gosec // No strong cryptography needed.
 	"fmt"
 	"strings"
 )
@@ -42,6 +42,7 @@ func ChildName(parent, suffix string) string {
 		// If the suffix is longer than the longest allowed suffix, then
 		// we hash the whole combined string and use that as the suffix.
 		if head-len(suffix) <= 0 {
+			// nolint:gosec // No strong cryptography needed.
 			h := md5.Sum([]byte(parent + suffix))
 			// 1. trim parent, if needed
 			if head < len(parent) {
@@ -58,6 +59,7 @@ func ChildName(parent, suffix string) string {
 			// remove it.
 			return strings.TrimRight(ret, "-")
 		}
+		// nolint:gosec // No strong cryptography needed.
 		n = fmt.Sprintf("%s%x", parent[:head-len(suffix)], md5.Sum([]byte(parent)))
 	}
 	return n + suffix
diff --git a/metrics/config.go b/metrics/config.go
index e62250fb0..fdf3cb855 100644
--- a/metrics/config.go
+++ b/metrics/config.go
@@ -28,7 +28,6 @@ import (
 	"time"
 
 	"go.opencensus.io/stats"
-	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"knative.dev/pkg/metrics/metricskey"
 )
@@ -174,7 +173,7 @@ func (mc *metricsConfig) record(ctx context.Context, mss []stats.Measurement, ro
 	return mc.recorder(ctx, mss, ros...)
 }
 
-func createMetricsConfig(ctx context.Context, ops ExporterOptions, logger *zap.SugaredLogger) (*metricsConfig, error) {
+func createMetricsConfig(ctx context.Context, ops ExporterOptions) (*metricsConfig, error) {
 	var mc metricsConfig
 
 	if ops.Domain == "" {
diff --git a/metrics/exporter.go b/metrics/exporter.go
index da5c510c9..a606c8ad2 100644
--- a/metrics/exporter.go
+++ b/metrics/exporter.go
@@ -135,7 +135,7 @@ func UpdateExporterFromConfigMapWithOpts(ctx context.Context, opts ExporterOptio
 // and updating the current exporter.
 func UpdateExporter(ctx context.Context, ops ExporterOptions, logger *zap.SugaredLogger) error {
 	// TODO(https://github.com/knative/pkg/issues/1273): check if ops.secrets is `nil` after new metrics plan lands
-	newConfig, err := createMetricsConfig(ctx, ops, logger)
+	newConfig, err := createMetricsConfig(ctx, ops)
 	if err != nil {
 		if getCurMetricsConfig() == nil {
 			// Fail the process if there doesn't exist an exporter.
diff --git a/metrics/opencensus_exporter.go b/metrics/opencensus_exporter.go
index c0a475250..403449e23 100644
--- a/metrics/opencensus_exporter.go
+++ b/metrics/opencensus_exporter.go
@@ -99,6 +99,7 @@ func getCredentials(component string, secret *corev1.Secret, logger *zap.Sugared
 		return nil
 	}
 	return credentials.NewTLS(&tls.Config{
+		MinVersion: tls.VersionTLS12,
 		GetClientCertificate: func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
 			cert, err := tls.X509KeyPair(secret.Data["client-cert.pem"], secret.Data["client-key.pem"])
 			if err != nil {
diff --git a/webhook/conversion.go b/webhook/conversion.go
index 73cddd7a4..a54a6cb8e 100644
--- a/webhook/conversion.go
+++ b/webhook/conversion.go
@@ -36,7 +36,7 @@ type ConversionController interface {
 	Convert(context.Context, *apixv1.ConversionRequest) *apixv1.ConversionResponse
 }
 
-func conversionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c ConversionController) http.HandlerFunc {
+func conversionHandler(rootLogger *zap.SugaredLogger, _ StatsReporter, c ConversionController) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		logger := rootLogger
 		logger.Infof("Webhook ServeHTTP request=%#v", r)
diff --git a/webhook/webhook.go b/webhook/webhook.go
index 8d628c5e3..c65c9f3c3 100644
--- a/webhook/webhook.go
+++ b/webhook/webhook.go
@@ -183,6 +183,7 @@ func (wh *Webhook) Run(stop <-chan struct{}) error {
 		Handler: drainer,
 		Addr:    fmt.Sprintf(":%d", wh.Options.Port),
 		TLSConfig: &tls.Config{
+			MinVersion: tls.VersionTLS12,
 			GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
 				secret, err := wh.secretlister.Secrets(system.Namespace()).Get(wh.Options.SecretName)
 				if err != nil {
diff --git a/websocket/connection.go b/websocket/connection.go
index 360280227..00feceb50 100644
--- a/websocket/connection.go
+++ b/websocket/connection.go
@@ -64,6 +64,9 @@ type ManagedConnection struct {
 	closeChan chan struct{}
 	closeOnce sync.Once
 
+	establishChan chan struct{}
+	establishOnce sync.Once
+
 	// Used to capture asynchronous processes to be waited
 	// on when shutting the connection down.
 	processingWg sync.WaitGroup
@@ -92,6 +95,25 @@ func NewDurableSendingConnection(target string, logger *zap.SugaredLogger) *Mana
 	return NewDurableConnection(target, nil, logger)
 }
 
+// NewDurableSendingConnectionGuaranteed creates a new websocket connection
+// that can only send messages to the endpoint it connects to. It returns
+// the connection if the connection can be established within the given
+// `duration`. Otherwise it returns the ErrConnectionNotEstablished error.
+//
+// The connection will continuously be kept alive and reconnected
+// in case of a loss of connectivity.
+func NewDurableSendingConnectionGuaranteed(target string, duration time.Duration, logger *zap.SugaredLogger) (*ManagedConnection, error) {
+	c := NewDurableConnection(target, nil, logger)
+
+	select {
+	case <-c.establishChan:
+		return c, nil
+	case <-time.After(duration):
+		c.Shutdown()
+		return nil, ErrConnectionNotEstablished
+	}
+}
+
 // NewDurableConnection creates a new websocket connection, that
 // passes incoming messages to the given message channel. It can also
 // send messages to the endpoint it connects to.
@@ -175,6 +197,7 @@ func newConnection(connFactory func() (rawConnection, error), messageChan chan [
 	conn := &ManagedConnection{
 		connectionFactory: connFactory,
 		closeChan:         make(chan struct{}),
+		establishChan:     make(chan struct{}),
 		messageChan:       messageChan,
 		connectionBackoff: wait.Backoff{
 			Duration: 100 * time.Millisecond,
@@ -211,6 +234,9 @@ func (c *ManagedConnection) connect() error {
 			defer c.connectionLock.Unlock()
 
 			c.connection = conn
+			c.establishOnce.Do(func() {
+				close(c.establishChan)
+			})
 			return true, nil
 		case <-c.closeChan:
 			return false, errShuttingDown
@@ -327,11 +353,3 @@ func (c *ManagedConnection) Shutdown() error {
 	c.processingWg.Wait()
 	return err
 }
-
-// IsEstablished returns true the websocket connection has been established.
-func (c *ManagedConnection) IsEstablished() bool {
-	c.connectionLock.RLock()
-	defer c.connectionLock.RUnlock()
-
-	return c.connection != nil
-}


Test patch:
diff --git a/.github/workflows/knative-go-test.yaml b/.github/workflows/knative-go-test.yaml
index f8dcfd88d..bfe6c6ad2 100644
--- a/.github/workflows/knative-go-test.yaml
+++ b/.github/workflows/knative-go-test.yaml
@@ -31,7 +31,7 @@ jobs:
     name: Unit Tests
     strategy:
       matrix:
-        go-version: [1.14.x]
+        go-version: [1.15.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
diff --git a/metrics/config_test.go b/metrics/config_test.go
index d6c1f5e05..f5683e065 100644
--- a/metrics/config_test.go
+++ b/metrics/config_test.go
@@ -536,7 +536,7 @@ func TestGetMetricsConfig(t *testing.T) {
 	ctx := context.Background()
 	for _, test := range errorTests {
 		t.Run(test.name, func(t *testing.T) {
-			_, err := createMetricsConfig(ctx, test.ops, TestLogger(t))
+			_, err := createMetricsConfig(ctx, test.ops)
 			if err == nil || err.Error() != test.expectedErr {
 				t.Errorf("Wanted err: %v, got: %v", test.expectedErr, err)
 			}
@@ -546,7 +546,7 @@ func TestGetMetricsConfig(t *testing.T) {
 	successTestsInit()
 	for _, test := range successTests {
 		t.Run(test.name, func(t *testing.T) {
-			mc, err := createMetricsConfig(ctx, test.ops, TestLogger(t))
+			mc, err := createMetricsConfig(ctx, test.ops)
 			if err != nil {
 				t.Errorf("Wanted valid config %v, got error %v", test.expectedConfig, err)
 			}
@@ -641,7 +641,7 @@ func TestGetMetricsConfig_fromEnv(t *testing.T) {
 			os.Setenv(test.varName, test.varValue)
 			defer os.Unsetenv(test.varName)
 
-			mc, err := createMetricsConfig(ctx, test.ops, TestLogger(t))
+			mc, err := createMetricsConfig(ctx, test.ops)
 			if err != nil {
 				t.Errorf("Wanted valid config %v, got error %v", test.expectedConfig, err)
 			}
@@ -656,7 +656,7 @@ func TestGetMetricsConfig_fromEnv(t *testing.T) {
 			os.Setenv(test.varName, test.varValue)
 			defer os.Unsetenv(test.varName)
 
-			mc, err := createMetricsConfig(ctx, test.ops, TestLogger(t))
+			mc, err := createMetricsConfig(ctx, test.ops)
 			if mc != nil {
 				t.Errorf("Wanted no config, got %v", mc)
 			}
@@ -674,7 +674,7 @@ func TestIsNewExporterRequiredFromNilConfig(t *testing.T) {
 
 	for _, test := range successTests {
 		t.Run(test.name, func(t *testing.T) {
-			mc, err := createMetricsConfig(ctx, test.ops, TestLogger(t))
+			mc, err := createMetricsConfig(ctx, test.ops)
 			if err != nil {
 				t.Errorf("Wanted valid config %v, got error %v", test.expectedConfig, err)
 			}
@@ -1052,7 +1052,7 @@ func TestStackdriverRecord(t *testing.T) {
 				Domain:    "knative.dev/internal/serving",
 				Component: "activator",
 			}
-			mc, err := createMetricsConfig(context.Background(), opts, TestLogger(t))
+			mc, err := createMetricsConfig(context.Background(), opts)
 			if err != nil {
 				t.Errorf("Expected valid config %+v, got error: %v\n", opts, err)
 			}
diff --git a/test/logstream/kubelogs.go b/test/logstream/kubelogs.go
index 592d3981d..ecd217329 100644
--- a/test/logstream/kubelogs.go
+++ b/test/logstream/kubelogs.go
@@ -110,7 +110,7 @@ func (k *kubelogs) watchPods(t test.TLegacy) {
 		t.Error("Logstream knative pod watch failed, logs might be missing", "error", err)
 		return
 	}
-	go func() error {
+	go func() {
 		watchedPods := sets.NewString()
 		for ev := range wi.ResultChan() {
 			p := ev.Object.(*corev1.Pod)
@@ -128,7 +128,6 @@ func (k *kubelogs) watchPods(t test.TLegacy) {
 				}
 			}
 		}
-		return nil
 	}()
 }
 
diff --git a/websocket/connection_test.go b/websocket/connection_test.go
index 69a674fe6..b98510187 100644
--- a/websocket/connection_test.go
+++ b/websocket/connection_test.go
@@ -419,16 +419,43 @@ func TestDurableConnectionSendsPingsRegularly(t *testing.T) {
 	}
 }
 
-func TestIsEstablished(t *testing.T) {
-	c := &ManagedConnection{}
-	if got, want := c.IsEstablished(), false; got != want {
-		t.Errorf("IsEstablished() = %v, want = %v", got, want)
+func TestNewDurableSendingConnectionGuaranteed(t *testing.T) {
+	// Unhappy case.
+	logger := ktesting.TestLogger(t)
+	_, err := NewDurableSendingConnectionGuaranteed("ws://somewhere.not.exist", time.Second, logger)
+	if got, want := err.Error(), ErrConnectionNotEstablished.Error(); got != want {
+		t.Errorf("Got error: %v, want error: %v", got, want)
 	}
 
-	c.connection = &inspectableConnection{
-		writeMessageCalls: make(chan struct{}, 1),
+	// Happy case.
+	const testPayload = "test"
+	reconnectChan := make(chan struct{})
+	upgrader := websocket.Upgrader{}
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+
+		// Waits for a message to be sent before dropping the connection.
+		<-reconnectChan
+		c.Close()
+	}))
+	defer s.Close()
+
+	target := "ws" + strings.TrimPrefix(s.URL, "http")
+	conn, err := NewDurableSendingConnectionGuaranteed(target, time.Second, logger)
+	if err != nil {
+		t.Errorf("Got error from NewDurableSendingConnectionGuaranteed: %v", err)
 	}
-	if got, want := c.IsEstablished(), true; got != want {
-		t.Errorf("IsEstablished() = %v, want = %v", got, want)
+	defer conn.Shutdown()
+
+	// Sending the message immediately should be fine as the connection has been established.
+	if err := conn.Send(testPayload); err != nil {
+		t.Errorf("Failed to send a message: %v", err)
 	}
+
+	// Message successfully sent, instruct the server to drop the connection.
+	reconnectChan <- struct{}{}
+
 }