Skip to content

Commit e15767a

Browse files
thegridmanthegridman
committed
Enh 37496193 - [37496162->24.09.2] General refactoring and hardening of the gRPC APIs
(merge ce/main -> ce/24.09 113641) [git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v24.09/": change = 113642]
1 parent e856cf5 commit e15767a

File tree

69 files changed

+2802
-365
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

69 files changed

+2802
-365
lines changed

‎prj/coherence-core-components/pom.xml

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
<?xml version="1.0" encoding="UTF-8"?>
22
<!--
3-
~ Copyright (c) 2000, 2025, Oracle and/or its affiliates.
4-
~
5-
~ Licensed under the Universal Permissive License v 1.0 as shown at
6-
~ https://oss.oracle.com/licenses/upl.
3+
Copyright (c) 2000, 2025, Oracle and/or its affiliates.
4+
5+
Licensed under the Universal Permissive License v 1.0 as shown at
6+
https://oss.oracle.com/licenses/upl.
77
-->
88

99
<project xmlns="http://maven.apache.org/POM/4.0.0"
@@ -70,6 +70,23 @@
7070
<optional>true</optional>
7171
</dependency>
7272

73+
<dependency>
74+
<groupId>io.grpc</groupId>
75+
<artifactId>grpc-stub</artifactId>
76+
<scope>provided</scope>
77+
<exclusions>
78+
<exclusion>
79+
<groupId>com.google.code.findbugs</groupId>
80+
<artifactId>jsr305</artifactId>
81+
</exclusion>
82+
</exclusions>
83+
</dependency>
84+
<dependency>
85+
<groupId>com.google.protobuf</groupId>
86+
<artifactId>protobuf-java</artifactId>
87+
<scope>provided</scope>
88+
</dependency>
89+
7390
</dependencies>
7491

7592
<build>

‎prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/extend/Connection.java

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11

22
/*
3-
* Copyright (c) 2000, 2023, Oracle and/or its affiliates.
3+
* Copyright (c) 2000, 2025, Oracle and/or its affiliates.
44
*
55
* Licensed under the Universal Permissive License v 1.0 as shown at
66
* https://oss.oracle.com/licenses/upl.
@@ -297,7 +297,19 @@ private com.tangosol.coherence.Component get_Module()
297297
{
298298
return this;
299299
}
300-
300+
301+
/**
302+
* Create a new instance of a {@link Channel}.
303+
*
304+
* @return a new instance of a {@link Channel}
305+
*/
306+
protected Channel createChannel()
307+
{
308+
Channel channel = new Channel();
309+
channel.setConnection(this);
310+
return channel;
311+
}
312+
301313
// From interface: com.tangosol.net.messaging.Connection
302314
public com.tangosol.net.messaging.Channel acceptChannel(java.net.URI uri, ClassLoader loader, com.tangosol.net.messaging.Channel.Receiver receiver, javax.security.auth.Subject subject)
303315
{
@@ -440,9 +452,8 @@ public com.tangosol.net.messaging.Channel acceptChannelResponse(int nId, com.tan
440452
throw new IllegalArgumentException("serializer cannot be null");
441453
}
442454

443-
Channel channel = new Channel();
455+
Channel channel = createChannel();
444456
channel.setId(nId);
445-
channel.setConnection(this);
446457
channel.setMessageFactory(factory);
447458
channel.setReceiver(receiver);
448459
channel.setSerializer(serializer);
@@ -684,9 +695,8 @@ public java.net.URI createChannelInternal(com.tangosol.net.messaging.Protocol pr
684695
int nId = generateChannelId();
685696

686697
// create a new Channel
687-
Channel channel = new Channel();
698+
Channel channel = createChannel();
688699
channel.setId(nId);
689-
channel.setConnection(this);
690700
channel.setReceiver(receiver);
691701
channel.setMessageFactory(factory);
692702
channel.setSerializer(serializer);
@@ -1170,8 +1180,7 @@ public void onInit()
11701180
// import com.tangosol.util.ThreadGate;
11711181

11721182
// create and register "Channel0"
1173-
Channel channel0 = new Channel();
1174-
channel0.setConnection(this);
1183+
Channel channel0 = createChannel();
11751184
registerChannel(channel0);
11761185

11771186
setThreadGate(new ThreadGate());
@@ -1306,8 +1315,7 @@ public int openChannelRequest(String sProtocol, com.tangosol.io.Serializer seria
13061315

13071316
int nId = generateChannelId();
13081317

1309-
Channel channel = new Channel();
1310-
channel.setConnection(this);
1318+
Channel channel = createChannel();
13111319
channel.setId(nId);
13121320
channel.setMessageFactory(factory);
13131321
channel.setReceiver(receiver);
@@ -1345,9 +1353,8 @@ public com.tangosol.net.messaging.Channel openChannelResponse(int nId, com.tango
13451353
throw new IllegalArgumentException("serializer cannot be null");
13461354
}
13471355

1348-
Channel channel = new Channel();
1356+
Channel channel = createChannel();
13491357
channel.setId(nId);
1350-
channel.setConnection(this);
13511358
channel.setMessageFactory(factory);
13521359
channel.setReceiver(receiver);
13531360
channel.setSerializer(serializer);

‎prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/extend/message/GrpcMessageWrapper.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/*
2+
* Copyright (c) 2000, 2025, Oracle and/or its affiliates.
3+
*
4+
* Licensed under the Universal Permissive License v 1.0 as shown at
5+
* https://oss.oracle.com/licenses/upl.
6+
*/
7+
8+
package com.tangosol.coherence.component.net.extend.message;
9+
10+
import com.google.protobuf.Any;
11+
12+
import com.tangosol.coherence.component.net.extend.message.response.GrpcResponse;
13+
import com.tangosol.io.Serializer;
14+
15+
import com.tangosol.net.messaging.Message;
16+
17+
/**
18+
* A {@link Message} implementation that wraps a protobuf message.
19+
*
20+
* @author Jonathan Knight 2025.01.25
21+
*/
22+
public interface GrpcMessageWrapper
23+
extends Message
24+
{
25+
/**
26+
* Set the wrapped protobuf message.
27+
*
28+
* @param any the wrapped protobuf message
29+
* @param serializer the serializer to deserialize binary payloads
30+
*/
31+
void setProtoMessage(Any any, Serializer serializer);
32+
33+
/**
34+
* Return the message response.
35+
*
36+
* @return the message response
37+
*/
38+
GrpcResponse getResponse();
39+
}

‎prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/extend/message/response/GrpcResponse.java

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
/*
2+
* Copyright (c) 2000, 2025, Oracle and/or its affiliates.
3+
*
4+
* Licensed under the Universal Permissive License v 1.0 as shown at
5+
* https://oss.oracle.com/licenses/upl.
6+
*/
7+
8+
package com.tangosol.coherence.component.net.extend.message.response;
9+
10+
import com.google.protobuf.Message;
11+
12+
import com.tangosol.coherence.component.net.extend.message.Response;
13+
14+
import com.tangosol.io.Serializer;
15+
16+
import io.grpc.stub.StreamObserver;
17+
18+
/**
19+
* A gRPC response message.
20+
*
21+
* @author Jonathan Knight 2025.01.25
22+
*/
23+
public abstract class GrpcResponse
24+
extends Response
25+
{
26+
/**
27+
* Default constructor.
28+
*/
29+
protected GrpcResponse()
30+
{
31+
super(null, null, true);
32+
}
33+
34+
/**
35+
* Set the proxy identifier for this message.
36+
*
37+
* @param proxyId the proxy identifier for this message
38+
*/
39+
public void setProxyId(int proxyId)
40+
{
41+
m_nProxyId = proxyId;
42+
}
43+
44+
/**
45+
* Return the proxy identifier for this message.
46+
*
47+
* @return the proxy identifier for this message
48+
*/
49+
public int getProxyId()
50+
{
51+
return m_nProxyId;
52+
}
53+
54+
/**
55+
* Set the {@link StreamObserver} to send responses to.
56+
*
57+
* @param observer the {@link StreamObserver} to send responses to
58+
*/
59+
public void setStreamObserver(StreamObserver<? extends Message> observer)
60+
{
61+
m_observer = observer;
62+
}
63+
64+
/**
65+
* Return the {@link StreamObserver} to send responses to.
66+
*
67+
* @return the {@link StreamObserver} to send responses to
68+
*/
69+
public StreamObserver<? extends Message> getStreamObserver()
70+
{
71+
return m_observer;
72+
}
73+
74+
/**
75+
* Set the {@link Serializer}.
76+
*
77+
* @param serializer the {@link Serializer}
78+
*/
79+
public void setSerializer(Serializer serializer)
80+
{
81+
m_serializer = serializer;
82+
}
83+
84+
/**
85+
* Return the {@link Serializer}.
86+
*
87+
* @return the {@link Serializer}
88+
*/
89+
public Serializer getSerializer()
90+
{
91+
return m_serializer;
92+
}
93+
94+
/**
95+
* Return {@code true} if the {@link StreamObserver} should be
96+
* completed after returning the response.
97+
*
98+
* @return {@code true} if the {@link StreamObserver} should be
99+
* completed after returning the response
100+
*/
101+
public boolean completeStream()
102+
{
103+
return true;
104+
}
105+
106+
/**
107+
* Create a protobuf response message.
108+
*
109+
* @return a protobuf response message
110+
*/
111+
public abstract Message getProtoResponse();
112+
113+
// ----- data members ---------------------------------------------------
114+
115+
/**
116+
* The proxy identifier.
117+
*/
118+
protected int m_nProxyId;
119+
120+
/**
121+
* The {@link Serializer}.
122+
*/
123+
protected Serializer m_serializer;
124+
125+
/**
126+
* The {@link StreamObserver} to send responses to.
127+
*/
128+
protected StreamObserver<? extends Message> m_observer;
129+
}

‎prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/extend/messageFactory/GrpcMessageFactory.java

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/*
2+
* Copyright (c) 2000, 2025, Oracle and/or its affiliates.
3+
*
4+
* Licensed under the Universal Permissive License v 1.0 as shown at
5+
* https://oss.oracle.com/licenses/upl.
6+
*/
7+
8+
package com.tangosol.coherence.component.net.extend.messageFactory;
9+
10+
import com.google.protobuf.Message;
11+
12+
import com.tangosol.coherence.component.net.extend.message.GrpcMessageWrapper;
13+
14+
import com.tangosol.coherence.component.net.extend.message.response.GrpcResponse;
15+
16+
import com.tangosol.io.Serializer;
17+
18+
import com.tangosol.net.messaging.Protocol;
19+
20+
/**
21+
* A {@link Protocol.MessageFactory} that can also produce messages
22+
* from Protobuf requests.
23+
*
24+
* @author Jonathan Knight 2025.01.25
25+
*/
26+
public interface GrpcMessageFactory<Req extends Message, Resp extends Message>
27+
extends Protocol.MessageFactory
28+
{
29+
/**
30+
* Create a {@link GrpcMessageWrapper} from a Protobuf {@link Message}.
31+
*
32+
* @param request the {@link Message} to create the {@link GrpcMessageWrapper} from
33+
* @param serializer the {@link Serializer} to use to deserialize binary payloads
34+
* @param <M> the expected type of the message to return
35+
*
36+
* @return a {@link GrpcMessageWrapper} created from a Protobuf {@link Message}
37+
*/
38+
<M extends GrpcMessageWrapper> M createRequestMessage(Req request, Serializer serializer);
39+
40+
/**
41+
* Create a response {@link Message} that will be wrapped in a proxy response
42+
* before being sent to the response stream observer.
43+
*
44+
* @param response the {@link GrpcResponse} to convert to a {@link Message}
45+
*
46+
* @return the response {@link Message}
47+
*/
48+
Resp createResponse(GrpcResponse response);
49+
50+
/**
51+
* Convert a Coherence {@link com.tangosol.net.messaging.Message}
52+
* into a corresponding Protobuf {@link Message}.
53+
*
54+
* @param message the Coherence message to convert
55+
* @param nProxyId the proxy identifier for the message
56+
*
57+
* @return the corresponding Protobuf {@link Message}
58+
*/
59+
Resp toProtoMessage(com.tangosol.net.messaging.Message message, int nProxyId);
60+
}

‎prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/extend/proxy/GrpcExtendProxy.java

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
/*
2+
* Copyright (c) 2000, 2025, Oracle and/or its affiliates.
3+
*
4+
* Licensed under the Universal Permissive License v 1.0 as shown at
5+
* https://oss.oracle.com/licenses/upl.
6+
*/
7+
8+
package com.tangosol.coherence.component.net.extend.proxy;
9+
10+
import com.google.protobuf.Message;
11+
12+
import com.tangosol.coherence.component.util.daemon.queueProcessor.service.peer.acceptor.grpcAcceptor.GrpcChannel;
13+
14+
import com.tangosol.net.messaging.Channel;
15+
16+
/**
17+
* A gRPC extend proxy.
18+
*
19+
* @author Jonathan Knight 2025.01.25
20+
*/
21+
public interface GrpcExtendProxy<Resp extends Message>
22+
extends Channel.Receiver
23+
{
24+
/**
25+
* Return the {@link GrpcChannel} used by this proxy.
26+
*
27+
* @return the {@link GrpcChannel} used by this proxy
28+
*/
29+
@SuppressWarnings("unchecked")
30+
default GrpcChannel<Resp> getGrpcChannel()
31+
{
32+
return (GrpcChannel<Resp>) getChannel();
33+
}
34+
35+
/**
36+
* Return the {@link Channel} used by this proxy.
37+
*
38+
* @return the {@link Channel} used by this proxy
39+
*/
40+
Channel getChannel();
41+
}

0 commit comments

Comments
 (0)