Repository: influxdata/influxdb-client-go (go)

Issue:
Unsynchronized WriteAPIBlocking.Write* functions
`WriteAPIBlocking.WritePoint` and `WriteAPIBlocking.WriteRecord` must ensure synchronized calling of  `write.Service.HandleWrite`


Pull request description:
fix: synchronizing access to write service
Closes #208

## Proposed Changes

Synchronizing access to the `write.Service` in the `WriteAPIBlocking.write` function to prevent race conditions.

## Checklist

<!-- Checkboxes below this note can be erased if not applicable to your Pull Request. -->

- [X] CHANGELOG.md updated
- [X] Rebased/mergeable
- [X] A test has been added if appropriate
- [X] Tests pass
- [X] Commit messages are in [semantic format](https://seesparkbox.com/foundry/semantic_commit_messages)


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6a1e0e3..1c244fc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 1. [#206](https://github.com/influxdata/influxdb-client-go/pull/206) Adding TasksAPI for managing tasks and associated logs and runs.
 
 ### Bug fixes
+1. [#209](https://github.com/influxdata/influxdb-client-go/pull/209) Synchronizing access to the write service in WriteAPIBlocking.
 
 ### Documentation
 
diff --git a/api/writeAPIBlocking.go b/api/writeAPIBlocking.go
index f73609b..d50aa48 100644
--- a/api/writeAPIBlocking.go
+++ b/api/writeAPIBlocking.go
@@ -7,6 +7,7 @@ package api
 import (
 	"context"
 	"strings"
+	"sync"
 
 	http2 "github.com/influxdata/influxdb-client-go/v2/api/http"
 	"github.com/influxdata/influxdb-client-go/v2/api/write"
@@ -60,6 +61,7 @@ type WriteAPIBlocking interface {
 type writeAPIBlocking struct {
 	service      *iwrite.Service
 	writeOptions *write.Options
+	lock         sync.Mutex
 }
 
 // NewWriteAPIBlocking creates new WriteAPIBlocking instance for org and bucket with underlying client
@@ -68,6 +70,8 @@ func NewWriteAPIBlocking(org string, bucket string, service http2.Service, write
 }
 
 func (w *writeAPIBlocking) write(ctx context.Context, line string) error {
+	w.lock.Lock()
+	defer w.lock.Unlock()
 	err := w.service.HandleWrite(ctx, iwrite.NewBatch(line, w.writeOptions.RetryInterval()))
 	return err
 }


Test patch:
diff --git a/api/writeAPIBlocking_test.go b/api/writeAPIBlocking_test.go
index fd46a8e..407903b 100644
--- a/api/writeAPIBlocking_test.go
+++ b/api/writeAPIBlocking_test.go
@@ -72,3 +72,30 @@ func TestWriteContextCancel(t *testing.T) {
 	require.Equal(t, context.Canceled, err)
 	assert.Len(t, service.Lines(), 0)
 }
+
+func TestWriteParallel(t *testing.T) {
+	service := test.NewTestService(t, "http://localhost:8888")
+	writeAPI := NewWriteAPIBlocking("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
+	lines := genRecords(1000)
+
+	chanLine := make(chan string)
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			for l := range chanLine {
+				err := writeAPI.WriteRecord(context.Background(), l)
+				assert.Nil(t, err)
+			}
+			wg.Done()
+		}()
+	}
+	for _, l := range lines {
+		chanLine <- l
+	}
+	close(chanLine)
+	wg.Wait()
+	assert.Len(t, service.Lines(), len(lines))
+
+	service.Close()
+}