Package: DefaultClient$Builder

DefaultClient$Builder

name	instruction	branch	complexity	line	method
DefaultClient.Builder(Channel)	M: 9 C: 0 0%	M: 0 C: 0 100%	M: 1 C: 0 0%	M: 4 C: 0 0%	M: 1 C: 0 0%
build()	M: 29 C: 0 0%	M: 2 C: 0 0%	M: 2 C: 0 0%	M: 11 C: 0 0%	M: 1 C: 0 0%
builder()	M: 2 C: 0 0%	M: 0 C: 0 100%	M: 1 C: 0 0%	M: 1 C: 0 0%	M: 1 C: 0 0%

Coverage

1: /*******************************************************************************
2: * Copyright (c) 2017, 2022 Red Hat Inc and others.
3: *
4: * This program and the accompanying materials are made
5: * available under the terms of the Eclipse Public License 2.0
6: * which is available at https://www.eclipse.org/legal/epl-2.0/
7: *
8: * SPDX-License-Identifier: EPL-2.0
9: *
10: * Contributors:
11: * Red Hat Inc - initial API and implementation
12: *******************************************************************************/
13: package org.eclipse.kapua.client.gateway.spi;
14:
15: import java.util.Collection;
16: import java.util.Objects;
17: import java.util.Optional;
18: import java.util.Set;
19: import java.util.concurrent.CompletionStage;
20: import java.util.concurrent.Executors;
21: import java.util.concurrent.ScheduledExecutorService;
22:
23: import org.eclipse.kapua.client.gateway.Client;
24: import org.eclipse.kapua.client.gateway.ErrorHandler;
25: import org.eclipse.kapua.client.gateway.MessageHandler;
26: import org.eclipse.kapua.client.gateway.Payload;
27: import org.eclipse.kapua.client.gateway.Topic;
28: import org.eclipse.kapua.client.gateway.spi.util.MoreExecutors;
29:
30: public class DefaultClient extends AbstractClient {
31:
32: public static final class Builder extends AbstractClient.Builder<Builder> {
33:
34: private final Channel channel;
35:
36: public Builder(final Channel channel) {
37: Objects.requireNonNull(channel);
38:
39: this.channel = channel;
40: }
41:
42: protected Builder builder() {
43: return this;
44: }
45:
46: @Override
47: public Client build() throws Exception {
48: ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
49:
50: try {
51: final DefaultClient client = new DefaultClient(channel, executor, modules());
52: try {
53: client.init();
54: } catch (Exception e) {
55: // init failed, close
56: client.close();
57: // and rethrow
58: throw e;
59: }
60:
61: // all good, claim instance
62: executor = null;
63:
64: // and return
65: return client;
66: } finally {
67:• if (executor != null) {
68: // not claime -> dispose
69: executor.shutdown();
70: }
71: }
72: }
73: }
74:
75: private final Channel.Context context = new Channel.Context() {
76:
77: @Override
78: public void notifyConnected() {
79: DefaultClient.this.notifyConnected();
80: }
81:
82: @Override
83: public void notifyDisconnected() {
84: DefaultClient.this.notifyDisconnected();
85: }
86:
87: @Override
88: public ScheduledExecutorService executor() {
89: return MoreExecutors.preventShutdown(executor);
90: }
91:
92: };
93:
94: private final Channel channel;
95:
96: public DefaultClient(final Channel channel, final ScheduledExecutorService executor, final Set<Module> modules) {
97: super(executor, modules);
98:
99: Objects.requireNonNull(channel);
100: this.channel = channel;
101: this.channel.handleInit(context);
102: }
103:
104: @Override
105: public void close() throws Exception {
106: channel.handleClose(context);
107: executor.shutdown();
108: }
109:
110: @Override
111: protected <T> Optional<T> adaptModuleContext(final Class<T> clazz) {
112:
113: final Optional<T> result = channel.adapt(clazz);
114: if (result.isPresent()) {
115: return result;
116: }
117:
118: return super.adaptModuleContext(clazz);
119: }
120:
121: @Override
122: protected CompletionStage<?> handleSubscribe(String applicationId, Topic topic, MessageHandler messageHandler, ErrorHandler<? extends Throwable> errorHandler) {
123: return channel.handleSubscribe(applicationId, topic, messageHandler, errorHandler);
124: }
125:
126: @Override
127: protected CompletionStage<?> handlePublish(String applicationId, Topic topic, Payload payload) {
128: return channel.handlePublish(applicationId, topic, payload);
129: }
130:
131: @Override
132: protected void handleUnsubscribe(String applicationId, Collection<Topic> topics) throws Exception {
133: channel.handleUnsubscribe(applicationId, topics);
134: }
135:
136: }