Skip to content

Commit 8df14e4

Browse files
mdeutchmdeutch
committed
Add a java SDK to all of Core's functionalities
Signed-off-by: Michal Deutch <mdeutch@salesforce.com>
1 parent c039416 commit 8df14e4

13 files changed

Lines changed: 1441 additions & 103 deletions

File tree

‎sdk/java/src/main/java/com/gojek/feast/FeastClient.java‎

Lines changed: 11 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -25,31 +25,19 @@
2525
import feast.proto.serving.ServingServiceGrpc;
2626
import feast.proto.serving.ServingServiceGrpc.ServingServiceBlockingStub;
2727
import io.grpc.CallCredentials;
28+
import io.grpc.Channel;
2829
import io.grpc.ManagedChannel;
29-
import io.grpc.ManagedChannelBuilder;
30-
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
31-
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
32-
import io.opentracing.contrib.grpc.TracingClientInterceptor;
33-
import io.opentracing.util.GlobalTracer;
34-
import java.io.File;
3530
import java.util.HashSet;
3631
import java.util.List;
3732
import java.util.Optional;
38-
import java.util.concurrent.TimeUnit;
3933
import java.util.stream.Collectors;
40-
import javax.net.ssl.SSLException;
4134
import org.slf4j.Logger;
4235
import org.slf4j.LoggerFactory;
4336

4437
@SuppressWarnings("WeakerAccess")
45-
public class FeastClient implements AutoCloseable {
38+
public class FeastClient extends GrpcManager<ServingServiceBlockingStub> {
4639
Logger logger = LoggerFactory.getLogger(FeastClient.class);
4740

48-
private static final int CHANNEL_SHUTDOWN_TIMEOUT_SEC = 5;
49-
50-
private final ManagedChannel channel;
51-
private final ServingServiceBlockingStub stub;
52-
5341
/**
5442
* Create a client to access Feast Serving.
5543
*
@@ -73,33 +61,7 @@ public static FeastClient create(String host, int port) {
7361
* @return {@link FeastClient}
7462
*/
7563
public static FeastClient createSecure(String host, int port, SecurityConfig securityConfig) {
76-
// Configure client TLS
77-
ManagedChannel channel = null;
78-
if (securityConfig.isTLSEnabled()) {
79-
if (securityConfig.getCertificatePath().isPresent()) {
80-
String certificatePath = securityConfig.getCertificatePath().get();
81-
// Use custom certificate for TLS
82-
File certificateFile = new File(certificatePath);
83-
try {
84-
channel =
85-
NettyChannelBuilder.forAddress(host, port)
86-
.useTransportSecurity()
87-
.sslContext(GrpcSslContexts.forClient().trustManager(certificateFile).build())
88-
.build();
89-
} catch (SSLException e) {
90-
throw new IllegalArgumentException(
91-
String.format("Invalid Certificate provided at path: %s", certificatePath), e);
92-
}
93-
} else {
94-
// Use system certificates for TLS
95-
channel = ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build();
96-
}
97-
} else {
98-
// Disable TLS
99-
channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
100-
}
101-
102-
return new FeastClient(channel, securityConfig.getCredentials());
64+
return new FeastClient(host, port, securityConfig);
10365
}
10466

10567
/**
@@ -188,24 +150,16 @@ public List<Row> getOnlineFeatures(List<String> featureRefs, List<Row> rows, Str
188150
.collect(Collectors.toList());
189151
}
190152

191-
protected FeastClient(ManagedChannel channel, Optional<CallCredentials> credentials) {
192-
this.channel = channel;
193-
TracingClientInterceptor tracingInterceptor =
194-
TracingClientInterceptor.newBuilder().withTracer(GlobalTracer.get()).build();
195-
196-
ServingServiceBlockingStub servingStub =
197-
ServingServiceGrpc.newBlockingStub(tracingInterceptor.intercept(channel));
198-
199-
if (credentials.isPresent()) {
200-
servingStub = servingStub.withCallCredentials(credentials.get());
201-
}
153+
@Override
154+
protected ServingServiceBlockingStub getStub(Channel channel) {
155+
return ServingServiceGrpc.newBlockingStub(channel);
156+
}
202157

203-
this.stub = servingStub;
158+
protected FeastClient(ManagedChannel channel, Optional<CallCredentials> credentials) {
159+
super(channel, credentials);
204160
}
205161

206-
public void close() throws Exception {
207-
if (channel != null) {
208-
channel.shutdown().awaitTermination(CHANNEL_SHUTDOWN_TIMEOUT_SEC, TimeUnit.SECONDS);
209-
}
162+
protected FeastClient(String host, int port, SecurityConfig securityConfig) {
163+
super(host, port, securityConfig);
210164
}
211165
}

‎sdk/java/src/main/java/com/gojek/feast/GrpcManager.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2021 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package com.gojek.feast;
18+
19+
import io.grpc.CallCredentials;
20+
import io.grpc.Channel;
21+
import io.grpc.ManagedChannel;
22+
import io.grpc.ManagedChannelBuilder;
23+
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
24+
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
25+
import io.opentracing.contrib.grpc.TracingClientInterceptor;
26+
import io.opentracing.util.GlobalTracer;
27+
import java.io.File;
28+
import java.util.Optional;
29+
import java.util.concurrent.TimeUnit;
30+
import javax.net.ssl.SSLException;
31+
32+
public abstract class GrpcManager<S extends io.grpc.stub.AbstractBlockingStub<S>>
33+
implements AutoCloseable {
34+
35+
private static final int CHANNEL_SHUTDOWN_TIMEOUT_SEC = 5;
36+
37+
private final ManagedChannel channel;
38+
protected final S stub;
39+
40+
protected GrpcManager(String host, int port, SecurityConfig securityConfig) {
41+
this(createSecureChannel(host, port, securityConfig), securityConfig.getCredentials());
42+
}
43+
44+
protected abstract S getStub(Channel channel);
45+
46+
protected GrpcManager(ManagedChannel channel, Optional<CallCredentials> credentials) {
47+
this.channel = channel;
48+
TracingClientInterceptor tracingInterceptor =
49+
TracingClientInterceptor.newBuilder().withTracer(GlobalTracer.get()).build();
50+
51+
S servingStub = getStub(tracingInterceptor.intercept(channel));
52+
53+
if (credentials.isPresent()) {
54+
servingStub = servingStub.withCallCredentials(credentials.get());
55+
}
56+
57+
this.stub = servingStub;
58+
}
59+
60+
private static ManagedChannel createSecureChannel(
61+
String host, int port, SecurityConfig securityConfig) {
62+
ManagedChannel channel;
63+
if (securityConfig.isTLSEnabled()) {
64+
if (securityConfig.getCertificatePath().isPresent()) {
65+
String certificatePath = securityConfig.getCertificatePath().get();
66+
// Use custom certificate for TLS
67+
File certificateFile = new File(certificatePath);
68+
try {
69+
channel =
70+
NettyChannelBuilder.forAddress(host, port)
71+
.useTransportSecurity()
72+
.sslContext(GrpcSslContexts.forClient().trustManager(certificateFile).build())
73+
.build();
74+
} catch (SSLException e) {
75+
throw new IllegalArgumentException(
76+
String.format("Invalid Certificate provided at path: %s", certificatePath), e);
77+
}
78+
} else {
79+
// Use system certificates for TLS
80+
channel = ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build();
81+
}
82+
} else {
83+
// Disable TLS
84+
channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
85+
}
86+
return channel;
87+
}
88+
89+
@Override
90+
public void close() throws Exception {
91+
if (channel != null) {
92+
channel.shutdown().awaitTermination(CHANNEL_SHUTDOWN_TIMEOUT_SEC, TimeUnit.SECONDS);
93+
}
94+
}
95+
}

0 commit comments

Comments
 (0)