Skip to content

Commit 8be42f1

Browse files
nigelsmallnigelsmall
authored
Merge pull request #234 from apcj/write-on-read-session
More friendly error when user tries to WRITE in a READ session.
2 parents a015866 + 95a32f2 commit 8be42f1

File tree

7 files changed

+557
-302
lines changed

7 files changed

+557
-302
lines changed

‎driver/src/main/java/org/neo4j/driver/internal/ClusterDriver.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -286,7 +286,7 @@ public Session session()
286286
@Override
287287
public Session session( final AccessMode mode )
288288
{
289-
return new ClusteredNetworkSession( acquireConnection( mode ),
289+
return new ClusteredNetworkSession( mode, acquireConnection( mode ),
290290
new ClusteredErrorHandler()
291291
{
292292
@Override
@@ -357,4 +357,4 @@ ConnectionPool connectionPool()
357357
return connections;
358358
}
359359

360-
}
360+
}

‎driver/src/main/java/org/neo4j/driver/internal/ClusteredNetworkSession.java

Lines changed: 48 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -21,21 +21,27 @@
2121

2222
import org.neo4j.driver.internal.net.BoltServerAddress;
2323
import org.neo4j.driver.internal.spi.Connection;
24+
import org.neo4j.driver.v1.AccessMode;
2425
import org.neo4j.driver.v1.Logger;
2526
import org.neo4j.driver.v1.Statement;
2627
import org.neo4j.driver.v1.StatementResult;
2728
import org.neo4j.driver.v1.exceptions.ClientException;
2829
import org.neo4j.driver.v1.exceptions.ConnectionFailureException;
30+
import org.neo4j.driver.v1.exceptions.Neo4jException;
2931
import org.neo4j.driver.v1.exceptions.SessionExpiredException;
3032

33+
import static java.lang.String.format;
34+
3135
public class ClusteredNetworkSession extends NetworkSession
3236
{
37+
private final AccessMode mode;
3338
private final ClusteredErrorHandler onError;
3439

35-
ClusteredNetworkSession( Connection connection,
40+
ClusteredNetworkSession( AccessMode mode, Connection connection,
3641
ClusteredErrorHandler onError, Logger logger )
3742
{
3843
super( connection, logger );
44+
this.mode = mode;
3945
this.onError = onError;
4046
}
4147

@@ -44,25 +50,15 @@ public StatementResult run( Statement statement )
4450
{
4551
try
4652
{
47-
return new ClusteredStatementResult( super.run( statement ), connection.address(), onError );
53+
return new ClusteredStatementResult( super.run( statement ), mode, connection.address(), onError );
4854
}
4955
catch ( ConnectionFailureException e )
5056
{
51-
onError.onConnectionFailure( connection.address() );
52-
throw new SessionExpiredException( "Failed to perform write load to server", e );
57+
throw sessionExpired( e, onError, connection.address() );
5358
}
5459
catch ( ClientException e )
5560
{
56-
if ( e.code().equals( "Neo.ClientError.Cluster.NotALeader" ) )
57-
{
58-
onError.onWriteFailure( connection.address() );
59-
throw new SessionExpiredException(
60-
String.format( "Server at %s no longer accepts writes", connection.address().toString() ) );
61-
}
62-
else
63-
{
64-
throw e;
65-
}
61+
throw filterFailureToWrite( e, mode, onError, connection.address() );
6662
}
6763
}
6864

@@ -75,10 +71,44 @@ public void close()
7571
}
7672
catch ( ConnectionFailureException e )
7773
{
78-
BoltServerAddress address = connection.address();
79-
onError.onConnectionFailure( address );
80-
throw new SessionExpiredException(
81-
String.format( "Server at %s is no longer available", address.toString() ), e );
74+
throw sessionExpired(e, onError, connection.address());
75+
}
76+
}
77+
78+
static Neo4jException filterFailureToWrite( ClientException e, AccessMode mode, ClusteredErrorHandler onError,
79+
BoltServerAddress address )
80+
{
81+
if ( isFailedToWrite( e ) )
82+
{
83+
// The server is unaware of the session mode, so we have to implement this logic in the driver.
84+
// In the future, we might be able to move this logic to the server.
85+
switch ( mode )
86+
{
87+
case READ:
88+
return new ClientException( "Write queries cannot be performed in READ access mode." );
89+
case WRITE:
90+
onError.onWriteFailure( address );
91+
return new SessionExpiredException( format( "Server at %s no longer accepts writes", address ) );
92+
default:
93+
throw new IllegalArgumentException( mode + " not supported." );
94+
}
8295
}
96+
else
97+
{
98+
return e;
99+
}
100+
}
101+
102+
static SessionExpiredException sessionExpired( ConnectionFailureException e, ClusteredErrorHandler onError,
103+
BoltServerAddress address )
104+
{
105+
onError.onConnectionFailure( address );
106+
return new SessionExpiredException( format( "Server at %s is no longer available", address.toString() ), e );
107+
}
108+
109+
private static boolean isFailedToWrite( ClientException e )
110+
{
111+
return e.code().equals( "Neo.ClientError.Cluster.NotALeader" ) ||
112+
e.code().equals( "Neo.ClientError.General.ForbiddenOnReadOnlyDatabase" );
83113
}
84114
}

‎driver/src/main/java/org/neo4j/driver/internal/ClusteredStatementResult.java

Lines changed: 28 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -21,24 +21,32 @@
2121
import java.util.List;
2222

2323
import org.neo4j.driver.internal.net.BoltServerAddress;
24+
import org.neo4j.driver.v1.AccessMode;
2425
import org.neo4j.driver.v1.Record;
2526
import org.neo4j.driver.v1.StatementResult;
2627
import org.neo4j.driver.v1.exceptions.ClientException;
2728
import org.neo4j.driver.v1.exceptions.ConnectionFailureException;
2829
import org.neo4j.driver.v1.exceptions.NoSuchRecordException;
29-
import org.neo4j.driver.v1.exceptions.SessionExpiredException;
3030
import org.neo4j.driver.v1.summary.ResultSummary;
3131
import org.neo4j.driver.v1.util.Function;
3232

33+
import static java.lang.String.format;
34+
35+
import static org.neo4j.driver.internal.ClusteredNetworkSession.filterFailureToWrite;
36+
import static org.neo4j.driver.internal.ClusteredNetworkSession.sessionExpired;
37+
3338
public class ClusteredStatementResult implements StatementResult
3439
{
3540
private final StatementResult delegate;
41+
private final AccessMode mode;
3642
private final BoltServerAddress address;
3743
private final ClusteredErrorHandler onError;
3844

39-
ClusteredStatementResult( StatementResult delegate, BoltServerAddress address, ClusteredErrorHandler onError )
45+
ClusteredStatementResult( StatementResult delegate, AccessMode mode, BoltServerAddress address,
46+
ClusteredErrorHandler onError )
4047
{
4148
this.delegate = delegate;
49+
this.mode = mode;
4250
this.address = address;
4351
this.onError = onError;
4452
}
@@ -52,18 +60,11 @@ public List<String> keys()
5260
}
5361
catch ( ConnectionFailureException e )
5462
{
55-
throw sessionExpired( e );
63+
throw sessionExpired( e, onError, address );
5664
}
5765
catch ( ClientException e )
5866
{
59-
if ( isFailedToWrite( e ) )
60-
{
61-
throw failedWrite();
62-
}
63-
else
64-
{
65-
throw e;
66-
}
67+
throw filterFailureToWrite( e, mode, onError, address );
6768
}
6869
}
6970

@@ -76,18 +77,11 @@ public boolean hasNext()
7677
}
7778
catch ( ConnectionFailureException e )
7879
{
79-
throw sessionExpired( e );
80+
throw sessionExpired( e, onError, address );
8081
}
8182
catch ( ClientException e )
8283
{
83-
if ( isFailedToWrite( e ) )
84-
{
85-
throw failedWrite();
86-
}
87-
else
88-
{
89-
throw e;
90-
}
84+
throw filterFailureToWrite( e, mode, onError, address );
9185
}
9286
}
9387

@@ -100,18 +94,11 @@ public Record next()
10094
}
10195
catch ( ConnectionFailureException e )
10296
{
103-
throw sessionExpired( e );
97+
throw sessionExpired( e, onError, address );
10498
}
10599
catch ( ClientException e )
106100
{
107-
if ( isFailedToWrite( e ) )
108-
{
109-
throw failedWrite();
110-
}
111-
else
112-
{
113-
throw e;
114-
}
101+
throw filterFailureToWrite( e, mode, onError, address );
115102
}
116103
}
117104

@@ -125,18 +112,11 @@ public Record single() throws NoSuchRecordException
125112
}
126113
catch ( ConnectionFailureException e )
127114
{
128-
throw sessionExpired( e );
115+
throw sessionExpired( e, onError, address );
129116
}
130117
catch ( ClientException e )
131118
{
132-
if ( isFailedToWrite( e ) )
133-
{
134-
throw failedWrite();
135-
}
136-
else
137-
{
138-
throw e;
139-
}
119+
throw filterFailureToWrite( e, mode, onError, address );
140120
}
141121
}
142122

@@ -149,18 +129,11 @@ public Record peek()
149129
}
150130
catch ( ConnectionFailureException e )
151131
{
152-
throw sessionExpired( e );
132+
throw sessionExpired( e, onError, address );
153133
}
154134
catch ( ClientException e )
155135
{
156-
if ( isFailedToWrite( e ) )
157-
{
158-
throw failedWrite();
159-
}
160-
else
161-
{
162-
throw e;
163-
}
136+
throw filterFailureToWrite( e, mode, onError, address );
164137
}
165138
}
166139

@@ -173,42 +146,28 @@ public List<Record> list()
173146
}
174147
catch ( ConnectionFailureException e )
175148
{
176-
throw sessionExpired( e );
149+
throw sessionExpired( e, onError, address );
177150
}
178151
catch ( ClientException e )
179152
{
180-
if ( isFailedToWrite( e ) )
181-
{
182-
throw failedWrite();
183-
}
184-
else
185-
{
186-
throw e;
187-
}
153+
throw filterFailureToWrite( e, mode, onError, address );
188154
}
189155
}
190156

191157
@Override
192-
public <T> List<T> list( Function<Record,T> mapFunction )
158+
public <T> List<T> list( Function<Record, T> mapFunction )
193159
{
194160
try
195161
{
196-
return delegate.list(mapFunction);
162+
return delegate.list( mapFunction );
197163
}
198164
catch ( ConnectionFailureException e )
199165
{
200-
throw sessionExpired( e );
166+
throw sessionExpired( e, onError, address );
201167
}
202168
catch ( ClientException e )
203169
{
204-
if ( isFailedToWrite( e ) )
205-
{
206-
throw failedWrite();
207-
}
208-
else
209-
{
210-
throw e;
211-
}
170+
throw filterFailureToWrite( e, mode, onError, address );
212171
}
213172
}
214173

@@ -227,35 +186,12 @@ public ResultSummary consume()
227186
}
228187
catch ( ConnectionFailureException e )
229188
{
230-
throw sessionExpired( e );
189+
throw sessionExpired( e, onError, address );
231190
}
232191
catch ( ClientException e )
233192
{
234-
if ( isFailedToWrite( e ) )
235-
{
236-
throw failedWrite();
237-
}
238-
else
239-
{
240-
throw e;
241-
}
193+
throw filterFailureToWrite( e, mode, onError, address );
242194
}
243195
}
244196

245-
private SessionExpiredException sessionExpired( ConnectionFailureException e )
246-
{
247-
onError.onConnectionFailure( address );
248-
return new SessionExpiredException( String.format( "Server at %s is no longer available", address.toString()), e);
249-
}
250-
251-
private SessionExpiredException failedWrite()
252-
{
253-
onError.onWriteFailure( address );
254-
return new SessionExpiredException( String.format( "Server at %s no longer accepts writes", address.toString()));
255-
}
256-
257-
private boolean isFailedToWrite( ClientException e )
258-
{
259-
return e.code().equals( "Neo.ClientError.Cluster.NotALeader" );
260-
}
261197
}

0 commit comments

Comments
 (0)