You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2022/01/25 14:19:22 UTC
[plc4x] branch develop updated: test(plc4go/connection-cache): Increased test-coverage for the Close functionality.
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new aa429e4 test(plc4go/connection-cache): Increased test-coverage for the Close functionality.
aa429e4 is described below
commit aa429e42438754c3fbb5c00b768218864f9ca711
Author: cdutz <ch...@c-ware.de>
AuthorDate: Tue Jan 25 15:17:49 2022 +0100
test(plc4go/connection-cache): Increased test-coverage for the Close functionality.
---
plc4go/pkg/plc4go/cache/plc_connection_cache.go | 62 ++++++++----------
.../pkg/plc4go/cache/plc_connection_cache_test.go | 74 +++++++++++++++++++++-
2 files changed, 99 insertions(+), 37 deletions(-)
diff --git a/plc4go/pkg/plc4go/cache/plc_connection_cache.go b/plc4go/pkg/plc4go/cache/plc_connection_cache.go
index 23ab7c0..894909c 100644
--- a/plc4go/pkg/plc4go/cache/plc_connection_cache.go
+++ b/plc4go/pkg/plc4go/cache/plc_connection_cache.go
@@ -163,28 +163,18 @@ func (t *plcConnectionCache) Close() <-chan PlcConnectionCacheCloseResult {
// We're just getting the lease as this way we can be sure nobody else is using it.
// We also really don't care if it worked, or not ... it's just an attempt of being
// nice.
- case connectionConnectResult := <-leaseResults:
- container.connection.Close()
+ case _ = <-leaseResults:
// Give back the connection.
- if connectionConnectResult.GetErr() == nil {
- connectionConnectResult.GetConnection().Close()
- }
+ container.connection.Close()
// If we're timing out brutally kill the connection.
case <-time.After(t.maxWaitTime):
// Forcefully close this connection.
container.connection.Close()
}
- // Check if all are closed and send the "all closed" event if all are closed.
- allClosed := true
- for _, cc2 := range t.connections {
- if !cc2.closed {
- allClosed = false
- break
- }
- }
- if allClosed {
- ch <- newDefaultPlcConnectionCacheCloseResult(t, nil)
+ select {
+ case ch <- newDefaultPlcConnectionCacheCloseResult(t, nil):
+ case <-time.After(time.Millisecond * 10):
}
}(cc)
}
@@ -254,7 +244,7 @@ func (t *connectionContainer) connect() {
// Mark the connection as being used.
t.state = StateInUse
// Return the lease to the caller.
- connection := newLeasedPlcConnection(t, t.leaseCounter, t.connection)
+ connection := newPlcConnectionLease(t, t.leaseCounter, t.connection)
// In this case we don't need to check for blocks
// as the getConnection function of the connection cache
// is definitely eagerly waiting for input.
@@ -297,7 +287,7 @@ func (t *connectionContainer) lease() <-chan plc4go.PlcConnectionConnectResult {
// Check if the connection is available.
if t.state == StateIdle {
t.leaseCounter++
- connection := newLeasedPlcConnection(t, t.leaseCounter, t.connection)
+ connection := newPlcConnectionLease(t, t.leaseCounter, t.connection)
// In this case we don't need to check for blocks
// as the getConnection function of the connection cache
// is definitely eagerly waiting for input.
@@ -328,7 +318,7 @@ func (t *connectionContainer) returnConnection(state cachedPlcConnectionState) e
next := t.queue[0]
t.queue = t.queue[1:]
t.leaseCounter++
- connection := newLeasedPlcConnection(t, t.leaseCounter, t.connection)
+ connection := newPlcConnectionLease(t, t.leaseCounter, t.connection)
// Send asynchronously as the receiver might have given up waiting,
// and we don't want anything to block here. 1ms should be enough for
// the calling process to reach the blocking read.
@@ -346,10 +336,10 @@ func (t *connectionContainer) returnConnection(state cachedPlcConnectionState) e
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-// leasedPlcConnection
+// plcConnectionLease
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-type leasedPlcConnection struct {
+type plcConnectionLease struct {
// Reference back to the container, so we can give the connection back.
connectionContainer *connectionContainer
// Counter for the number of times this connection has been used before.
@@ -358,8 +348,8 @@ type leasedPlcConnection struct {
connection spi.PlcConnection
}
-func newLeasedPlcConnection(connectionContainer *connectionContainer, leaseId uint32, connection spi.PlcConnection) *leasedPlcConnection {
- p := &leasedPlcConnection{
+func newPlcConnectionLease(connectionContainer *connectionContainer, leaseId uint32, connection spi.PlcConnection) *plcConnectionLease {
+ p := &plcConnectionLease{
connectionContainer: connectionContainer,
leaseId: leaseId,
connection: connection,
@@ -370,32 +360,32 @@ func newLeasedPlcConnection(connectionContainer *connectionContainer, leaseId ui
return p
}
-func (t *leasedPlcConnection) IsTraceEnabled() bool {
+func (t *plcConnectionLease) IsTraceEnabled() bool {
if t.connection == nil {
panic("Called 'IsTraceEnabled' on a closed cached connection")
}
return t.connection.IsTraceEnabled()
}
-func (t *leasedPlcConnection) GetTracer() *spi.Tracer {
+func (t *plcConnectionLease) GetTracer() *spi.Tracer {
if t.connection == nil {
panic("Called 'GetTracer' on a closed cached connection")
}
return t.connection.GetTracer()
}
-func (t *leasedPlcConnection) GetConnectionId() string {
+func (t *plcConnectionLease) GetConnectionId() string {
if t.connection == nil {
panic("Called 'GetConnectionId' on a closed cached connection")
}
return fmt.Sprintf("%s-%d", t.connection.GetConnectionId(), t.leaseId)
}
-func (t *leasedPlcConnection) Connect() <-chan plc4go.PlcConnectionConnectResult {
+func (t *plcConnectionLease) Connect() <-chan plc4go.PlcConnectionConnectResult {
panic("Called 'Connect' on a cached connection")
}
-func (t *leasedPlcConnection) BlockingClose() {
+func (t *plcConnectionLease) BlockingClose() {
if t.connection == nil {
panic("Called 'BlockingClose' on a closed cached connection")
}
@@ -403,7 +393,7 @@ func (t *leasedPlcConnection) BlockingClose() {
<-t.Close()
}
-func (t *leasedPlcConnection) Close() <-chan plc4go.PlcConnectionCloseResult {
+func (t *plcConnectionLease) Close() <-chan plc4go.PlcConnectionCloseResult {
if t.connection == nil {
panic("Called 'Close' on a closed cached connection")
}
@@ -458,56 +448,56 @@ func (t *leasedPlcConnection) Close() <-chan plc4go.PlcConnectionCloseResult {
return result
}
-func (t *leasedPlcConnection) IsConnected() bool {
+func (t *plcConnectionLease) IsConnected() bool {
if t.connection == nil {
return false
}
return t.connection.IsConnected()
}
-func (t *leasedPlcConnection) Ping() <-chan plc4go.PlcConnectionPingResult {
+func (t *plcConnectionLease) Ping() <-chan plc4go.PlcConnectionPingResult {
if t.connection == nil {
panic("Called 'Ping' on a closed cached connection")
}
return t.connection.Ping()
}
-func (t *leasedPlcConnection) GetMetadata() model.PlcConnectionMetadata {
+func (t *plcConnectionLease) GetMetadata() model.PlcConnectionMetadata {
if t.connection == nil {
panic("Called 'GetMetadata' on a closed cached connection")
}
return t.connection.GetMetadata()
}
-func (t *leasedPlcConnection) ReadRequestBuilder() model.PlcReadRequestBuilder {
+func (t *plcConnectionLease) ReadRequestBuilder() model.PlcReadRequestBuilder {
if t.connection == nil {
panic("Called 'ReadRequestBuilder' on a closed cached connection")
}
return t.connection.ReadRequestBuilder()
}
-func (t *leasedPlcConnection) WriteRequestBuilder() model.PlcWriteRequestBuilder {
+func (t *plcConnectionLease) WriteRequestBuilder() model.PlcWriteRequestBuilder {
if t.connection == nil {
panic("Called 'WriteRequestBuilder' on a closed cached connection")
}
return t.connection.WriteRequestBuilder()
}
-func (t *leasedPlcConnection) SubscriptionRequestBuilder() model.PlcSubscriptionRequestBuilder {
+func (t *plcConnectionLease) SubscriptionRequestBuilder() model.PlcSubscriptionRequestBuilder {
if t.connection == nil {
panic("Called 'SubscriptionRequestBuilder' on a closed cached connection")
}
return t.connection.SubscriptionRequestBuilder()
}
-func (t *leasedPlcConnection) UnsubscriptionRequestBuilder() model.PlcUnsubscriptionRequestBuilder {
+func (t *plcConnectionLease) UnsubscriptionRequestBuilder() model.PlcUnsubscriptionRequestBuilder {
if t.connection == nil {
panic("Called 'UnsubscriptionRequestBuilder' on a closed cached connection")
}
return t.connection.UnsubscriptionRequestBuilder()
}
-func (t *leasedPlcConnection) BrowseRequestBuilder() model.PlcBrowseRequestBuilder {
+func (t *plcConnectionLease) BrowseRequestBuilder() model.PlcBrowseRequestBuilder {
if t.connection == nil {
panic("Called 'BrowseRequestBuilder' on a closed cached connection")
}
diff --git a/plc4go/pkg/plc4go/cache/plc_connection_cache_test.go b/plc4go/pkg/plc4go/cache/plc_connection_cache_test.go
index 933c708..d90d037 100644
--- a/plc4go/pkg/plc4go/cache/plc_connection_cache_test.go
+++ b/plc4go/pkg/plc4go/cache/plc_connection_cache_test.go
@@ -93,6 +93,78 @@ func TestPlcConnectionCache_GetConnection(t1 *testing.T) {
}
}
+func TestPlcConnectionCache_Close(t1 *testing.T) {
+ type fields struct {
+ driverManager plc4go.PlcDriverManager
+ }
+ type args struct {
+ connectionStrings []string
+ }
+ tests := []struct {
+ name string
+ fields fields
+ args args
+ wantErr bool
+ wantTimeout bool
+ }{
+ {name: "simple",
+ fields: fields{
+ driverManager: func() plc4go.PlcDriverManager {
+ driverManager := plc4go.NewPlcDriverManager()
+ driverManager.RegisterDriver(simulated.NewDriver())
+ return driverManager
+ }(),
+ }, args: args{
+ connectionStrings: []string{
+ "simulated://1.2.3.4:42",
+ "simulated://4.3.2.1:23",
+ "simulated://0.8.1.15:7",
+ },
+ },
+ wantErr: false,
+ wantTimeout: false,
+ },
+ }
+ for _, tt := range tests {
+ t1.Run(tt.name, func(t1 *testing.T) {
+ cc := NewPlcConnectionCache(tt.fields.driverManager)
+ // Connect to all sources first
+ for _, connectionString := range tt.args.connectionStrings {
+ got := cc.GetConnection(connectionString)
+ select {
+ case connectResult := <-got:
+ if connectResult.GetErr() != nil {
+ t1.Errorf("PlcConnectionCache.GetConnection() error = %v", connectResult.GetErr())
+ } else {
+ // Give the connection back.
+ connectResult.GetConnection().Close()
+ }
+ case <-time.After(10 * time.Second):
+ if !tt.wantTimeout {
+ t1.Errorf("PlcConnectionCache.GetConnection() got timeout")
+ }
+ }
+ }
+ // Close all connections.
+ cacheCloseResults := cc.Close()
+ // Wait for all connections to be closed.
+ select {
+ case cacheCloseResult := <-cacheCloseResults:
+ if tt.wantErr && (cacheCloseResult.GetErr() == nil) {
+ t1.Errorf("PlcConnectionCache.Close() = %v, wantErr %v", cacheCloseResult.GetErr(), tt.wantErr)
+ } else if cacheCloseResult.GetErr() != nil {
+ t1.Errorf("PlcConnectionCache.Close() error = %v, wantErr %v", cacheCloseResult.GetErr(), tt.wantErr)
+ }
+ case <-time.After(10 * time.Second):
+ if !tt.wantTimeout {
+ t1.Errorf("PlcConnectionCache.Close() got timeout")
+ }
+ }
+
+ })
+ }
+}
+
func readFromPlc(t1 *testing.T, cache plcConnectionCache, connectionString string, resourceString string) <-chan []spi.TraceEntry {
ch := make(chan []spi.TraceEntry)
@@ -356,7 +428,7 @@ func TestPlcConnectionCache_ReturningConnectionWithPingError(t1 *testing.T) {
if connectResult.GetErr() != nil {
t1.Errorf("PlcConnectionCache.GetConnection() error = %v", connectResult.GetErr())
}
- connection := connectResult.GetConnection().(*leasedPlcConnection)
+ connection := connectResult.GetConnection().(*plcConnectionLease)
if connection != nil {
connectionCloseResultChan := connection.Close()
closeResult := <-connectionCloseResultChan