홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Shravan Ambati
Committed by Gerrit Code Review
5a11e17c 1 parent 18f3bce5

Refactored Kafka Application to simplify dependencies 

1. Fixed a Bug in KafkaProducer. Without this fix the App will not send data in GPB format.
2. Added two new services - KafkaProducerService and KafkaConfigService.
3. Fixed a TODO in the register API to return Kafka server information.
4. Removed the use of LeadershipService and ClusterService, since we are not ready for clustering yet.

Change-Id: If20ef5238bb4629af0c6769129494eb44abf1d3c
Expand all
Inline Side-by-side
Showing 18 changed files with 641 additions and 336 deletions
...@@ -39,6 +39,12 @@ ...@@ -39,6 +39,12 @@
39 <version>3.0.0-beta-2</version> 39 <version>3.0.0-beta-2</version>
40 </dependency> 40 </dependency>
41 41
42 + <dependency>
43 + <groupId>org.apache.servicemix.bundles</groupId>
44 + <artifactId>org.apache.servicemix.bundles.kafka-clients</artifactId>
45 + <version>0.8.2.2_1</version>
46 + </dependency>
47 +
42 </dependencies> 48 </dependencies>
43 49
44 <build> 50 <build>
......
...@@ -14,16 +14,16 @@ ...@@ -14,16 +14,16 @@
14 */ 14 */
15 package org.onosproject.kafkaintegration.api; 15 package org.onosproject.kafkaintegration.api;
16 16
17 +import java.util.List;
18 +
17 import org.onosproject.kafkaintegration.api.dto.EventSubscriber; 19 import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
18 -import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
19 import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type; 20 import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
21 +import org.onosproject.kafkaintegration.api.dto.RegistrationResponse;
20 import org.onosproject.kafkaintegration.errors.InvalidApplicationException; 22 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
21 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException; 23 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
22 24
23 import com.google.common.annotations.Beta; 25 import com.google.common.annotations.Beta;
24 26
25 -import java.util.List;
26 -
27 /** 27 /**
28 * APIs for subscribing to Onos Event Messages. 28 * APIs for subscribing to Onos Event Messages.
29 */ 29 */
...@@ -34,9 +34,9 @@ public interface EventSubscriptionService { ...@@ -34,9 +34,9 @@ public interface EventSubscriptionService {
34 * Registers the external application to receive events generated in ONOS. 34 * Registers the external application to receive events generated in ONOS.
35 * 35 *
36 * @param appName Application Name 36 * @param appName Application Name
37 - * @return unique consumer group identifier 37 + * @return Registration Response DTO.
38 */ 38 */
39 - EventSubscriberGroupId registerListener(String appName); 39 + RegistrationResponse registerListener(String appName);
40 40
41 /** 41 /**
42 * Removes the Registered Listener. 42 * Removes the Registered Listener.
......
1 +/**
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.kafkaintegration.api;
17 +
18 +import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
19 +
20 +public interface KafkaConfigService {
21 +
22 + /**
23 + * Returns the Kafka Server Configuration Information.
24 + *
25 + * @return KafkaServerConfig DTO Object.
26 + */
27 + KafkaServerConfig getConfigParams();
28 +
29 +}
30 +
1 +/**
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.kafkaintegration.api;
17 +
18 +import java.util.concurrent.Future;
19 +
20 +import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
21 +import org.apache.kafka.clients.producer.ProducerRecord;
22 +import org.apache.kafka.clients.producer.RecordMetadata;
23 +/**
24 + * APIs for controlling the Kafka Producer.
25 + *
26 + */
27 +public interface KafkaProducerService {
28 +
29 + /**
30 + * Starts the Kafka Producer.
31 + *
32 + * @param config the Kafka Server Config
33 + */
34 + void start(KafkaServerConfig config);
35 +
36 + /**
37 + * Stops the Kafka Producer.
38 + *
39 + */
40 + void stop();
41 +
42 + /**
43 + * Restarts the Kafka Producer.
44 + *
45 + * @param config the Kafka Server Config
46 + */
47 + void restart(KafkaServerConfig config);
48 +
49 + /**
50 + * Sends message to Kafka Server.
51 + *
52 + * @param record a message to be sent
53 + * @return metadata for a record that as been acknowledged
54 + */
55 + public Future<RecordMetadata> send(ProducerRecord<String, byte[]> record);
56 +}
...@@ -84,6 +84,7 @@ public final class DefaultEventSubscriber implements EventSubscriber { ...@@ -84,6 +84,7 @@ public final class DefaultEventSubscriber implements EventSubscriber {
84 .addValue(subscriberGroupId.toString()) 84 .addValue(subscriberGroupId.toString())
85 .add("eventType", eventType).toString(); 85 .add("eventType", eventType).toString();
86 } 86 }
87 +
87 /** 88 /**
88 * To create an instance of the builder. 89 * To create an instance of the builder.
89 * 90 *
...@@ -92,6 +93,7 @@ public final class DefaultEventSubscriber implements EventSubscriber { ...@@ -92,6 +93,7 @@ public final class DefaultEventSubscriber implements EventSubscriber {
92 public static Builder builder() { 93 public static Builder builder() {
93 return new Builder(); 94 return new Builder();
94 } 95 }
96 +
95 /** 97 /**
96 * Builder class for Event subscriber. 98 * Builder class for Event subscriber.
97 */ 99 */
...@@ -107,8 +109,7 @@ public final class DefaultEventSubscriber implements EventSubscriber { ...@@ -107,8 +109,7 @@ public final class DefaultEventSubscriber implements EventSubscriber {
107 } 109 }
108 110
109 @Override 111 @Override
110 - public Builder setSubscriberGroupId(EventSubscriberGroupId 112 + public Builder setSubscriberGroupId(EventSubscriberGroupId subscriberGroupId) {
111 - subscriberGroupId) {
112 this.subscriberGroupId = subscriberGroupId; 113 this.subscriberGroupId = subscriberGroupId;
113 return this; 114 return this;
114 } 115 }
...@@ -122,13 +123,11 @@ public final class DefaultEventSubscriber implements EventSubscriber { ...@@ -122,13 +123,11 @@ public final class DefaultEventSubscriber implements EventSubscriber {
122 @Override 123 @Override
123 public EventSubscriber build() { 124 public EventSubscriber build() {
124 checkNotNull(appName, "App name cannot be null"); 125 checkNotNull(appName, "App name cannot be null");
125 - checkNotNull(subscriberGroupId, "Subscriber group ID cannot " + 126 + checkNotNull(subscriberGroupId,
126 - "be " + 127 + "Subscriber group ID cannot " + "be " + "null");
127 - "null");
128 checkNotNull(eventType, "Event type cannot be null"); 128 checkNotNull(eventType, "Event type cannot be null");
129 129
130 - return new DefaultEventSubscriber(appName, 130 + return new DefaultEventSubscriber(appName, subscriberGroupId,
131 - subscriberGroupId,
132 eventType); 131 eventType);
133 } 132 }
134 } 133 }
......
1 +/**
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.kafkaintegration.api.dto;
17 +
18 +import static com.google.common.base.Preconditions.checkNotNull;
19 +
20 +/**
21 + * DTO to hold Kafka Server Configuration information.
22 + *
23 + */
24 +public final class KafkaServerConfig {
25 +
26 + private final String ipAddress;
27 +
28 + private final String port;
29 +
30 + private final int numOfRetries;
31 +
32 + private final int maxInFlightRequestsPerConnection;
33 +
34 + private final int acksRequired;
35 +
36 + private final String keySerializer;
37 +
38 + private final String valueSerializer;
39 +
40 + private KafkaServerConfig(String ipAddress, String port, int numOfRetries,
41 + int maxInFlightRequestsPerConnection,
42 + int requestRequiredAcks, String keySerializer,
43 + String valueSerializer) {
44 +
45 + this.ipAddress = checkNotNull(ipAddress, "Ip Address Cannot be null");
46 + this.port = checkNotNull(port, "Port Number cannot be null");
47 + this.numOfRetries = numOfRetries;
48 + this.maxInFlightRequestsPerConnection =
49 + maxInFlightRequestsPerConnection;
50 + this.acksRequired = requestRequiredAcks;
51 + this.keySerializer = keySerializer;
52 + this.valueSerializer = valueSerializer;
53 + }
54 +
55 + public final String getIpAddress() {
56 + return ipAddress;
57 + }
58 +
59 + public final String getPort() {
60 + return port;
61 + }
62 +
63 + public final int getNumOfRetries() {
64 + return numOfRetries;
65 + }
66 +
67 + public final int getMaxInFlightRequestsPerConnection() {
68 + return maxInFlightRequestsPerConnection;
69 + }
70 +
71 + public final int getAcksRequired() {
72 + return acksRequired;
73 + }
74 +
75 + public final String getKeySerializer() {
76 + return keySerializer;
77 + }
78 +
79 + public final String getValueSerializer() {
80 + return valueSerializer;
81 + }
82 +
83 + /**
84 + * To create an instance of the builder.
85 + *
86 + * @return instance of builder
87 + */
88 + public static Builder builder() {
89 + return new Builder();
90 + }
91 +
92 + /**
93 + * Builder class for KafkaServerConfig.
94 + */
95 + public static final class Builder {
96 + private String ipAddress;
97 +
98 + private String port;
99 +
100 + private int numOfRetries;
101 +
102 + private int maxInFlightRequestsPerConnection;
103 +
104 + private int acksRequired;
105 +
106 + private String keySerializer;
107 +
108 + private String valueSerializer;
109 +
110 + public Builder ipAddress(String ipAddress) {
111 + this.ipAddress = ipAddress;
112 + return this;
113 + }
114 +
115 + public Builder port(String port) {
116 + this.port = port;
117 + return this;
118 + }
119 +
120 + public Builder numOfRetries(int numOfRetries) {
121 + this.numOfRetries = numOfRetries;
122 + return this;
123 + }
124 +
125 + public Builder maxInFlightRequestsPerConnection(int maxInFlightRequestsPerConnection) {
126 + this.maxInFlightRequestsPerConnection =
127 + maxInFlightRequestsPerConnection;
128 + return this;
129 + }
130 +
131 + public Builder acksRequired(int acksRequired) {
132 + this.acksRequired = acksRequired;
133 + return this;
134 + }
135 +
136 + public Builder keySerializer(String keySerializer) {
137 + this.keySerializer = keySerializer;
138 + return this;
139 + }
140 +
141 + public Builder valueSerializer(String valueSerializer) {
142 + this.valueSerializer = valueSerializer;
143 + return this;
144 + }
145 +
146 + public KafkaServerConfig build() {
147 + checkNotNull(ipAddress, "App name cannot be null");
148 + checkNotNull(port, "Subscriber group ID cannot " + "be " + "null");
149 +
150 + return new KafkaServerConfig(ipAddress, port, numOfRetries,
151 + maxInFlightRequestsPerConnection,
152 + acksRequired, keySerializer,
153 + valueSerializer);
154 + }
155 + }
156 +}
1 +/**
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.kafkaintegration.api.dto;
17 +
18 +import static com.google.common.base.MoreObjects.toStringHelper;
19 +
20 +import java.util.Objects;
21 +
22 +/**
23 + * DTO to hold Registration Response for requests from external apps.
24 + */
25 +public final class RegistrationResponse {
26 +
27 + private EventSubscriberGroupId groupId;
28 +
29 + private String ipAddress;
30 +
31 + private String port;
32 +
33 + public RegistrationResponse(EventSubscriberGroupId groupId,
34 + String ipAddress, String port) {
35 + this.groupId = groupId;
36 + this.ipAddress = ipAddress;
37 + this.port = port;
38 + }
39 +
40 + public final EventSubscriberGroupId getGroupId() {
41 + return groupId;
42 + }
43 +
44 + public final String getIpAddress() {
45 + return ipAddress;
46 + }
47 +
48 + public final String getPort() {
49 + return port;
50 + }
51 +
52 + @Override
53 + public boolean equals(Object o) {
54 + if (o instanceof RegistrationResponse) {
55 + RegistrationResponse sub = (RegistrationResponse) o;
56 + if (sub.groupId.equals(groupId) && sub.ipAddress.equals(ipAddress)
57 + && sub.port.equals(port)) {
58 + return true;
59 + }
60 + }
61 +
62 + return false;
63 + }
64 +
65 + @Override
66 + public int hashCode() {
67 + return Objects.hash(groupId, ipAddress, port);
68 + }
69 +
70 + @Override
71 + public String toString() {
72 + return toStringHelper(this).add("subscriberGroupId", groupId)
73 + .add("ipAddress", ipAddress).add("port", port).toString();
74 + }
75 +}
...@@ -30,7 +30,8 @@ ...@@ -30,7 +30,8 @@
30 <packaging>bundle</packaging> 30 <packaging>bundle</packaging>
31 <description> 31 <description>
32 Kafka Integration Application. 32 Kafka Integration Application.
33 - This module is exclusive of REST calls and is only for the implementation of Apache Kafka. 33 + This module is exclusive of REST calls and is only for the implementation
34 + of the Application.
34 </description> 35 </description>
35 36
36 <dependencies> 37 <dependencies>
......
...@@ -15,7 +15,15 @@ ...@@ -15,7 +15,15 @@
15 */ 15 */
16 package org.onosproject.kafkaintegration.impl; 16 package org.onosproject.kafkaintegration.impl;
17 17
18 -import com.google.common.collect.ImmutableList; 18 +import static com.google.common.base.Preconditions.checkNotNull;
19 +import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
20 +import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
21 +
22 +import java.util.ArrayList;
23 +import java.util.List;
24 +import java.util.Map;
25 +import java.util.UUID;
26 +
19 import org.apache.felix.scr.annotations.Activate; 27 import org.apache.felix.scr.annotations.Activate;
20 import org.apache.felix.scr.annotations.Component; 28 import org.apache.felix.scr.annotations.Component;
21 import org.apache.felix.scr.annotations.Deactivate; 29 import org.apache.felix.scr.annotations.Deactivate;
...@@ -25,14 +33,15 @@ import org.apache.felix.scr.annotations.Service; ...@@ -25,14 +33,15 @@ import org.apache.felix.scr.annotations.Service;
25 import org.onosproject.core.ApplicationId; 33 import org.onosproject.core.ApplicationId;
26 import org.onosproject.core.CoreService; 34 import org.onosproject.core.CoreService;
27 import org.onosproject.kafkaintegration.api.EventSubscriptionService; 35 import org.onosproject.kafkaintegration.api.EventSubscriptionService;
36 +import org.onosproject.kafkaintegration.api.KafkaConfigService;
28 import org.onosproject.kafkaintegration.api.dto.DefaultEventSubscriber; 37 import org.onosproject.kafkaintegration.api.dto.DefaultEventSubscriber;
29 import org.onosproject.kafkaintegration.api.dto.EventSubscriber; 38 import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
30 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId; 39 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
40 +import org.onosproject.kafkaintegration.api.dto.RegistrationResponse;
31 import org.onosproject.kafkaintegration.api.dto.OnosEvent; 41 import org.onosproject.kafkaintegration.api.dto.OnosEvent;
32 import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type; 42 import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
33 import org.onosproject.kafkaintegration.errors.InvalidApplicationException; 43 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
34 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException; 44 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
35 -import org.onosproject.kafkaintegration.listener.ListenerFactory;
36 import org.onosproject.kafkaintegration.listener.OnosEventListener; 45 import org.onosproject.kafkaintegration.listener.OnosEventListener;
37 import org.onosproject.net.device.DeviceService; 46 import org.onosproject.net.device.DeviceService;
38 import org.onosproject.net.link.LinkService; 47 import org.onosproject.net.link.LinkService;
...@@ -42,14 +51,7 @@ import org.onosproject.store.service.StorageService; ...@@ -42,14 +51,7 @@ import org.onosproject.store.service.StorageService;
42 import org.slf4j.Logger; 51 import org.slf4j.Logger;
43 import org.slf4j.LoggerFactory; 52 import org.slf4j.LoggerFactory;
44 53
45 -import java.util.ArrayList; 54 +import com.google.common.collect.ImmutableList;
46 -import java.util.List;
47 -import java.util.Map;
48 -import java.util.UUID;
49 -
50 -import static com.google.common.base.Preconditions.checkNotNull;
51 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
52 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
53 55
54 /** 56 /**
55 * Implementation of Event Subscription Manager. 57 * Implementation of Event Subscription Manager.
...@@ -82,6 +84,9 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -82,6 +84,9 @@ public class EventSubscriptionManager implements EventSubscriptionService {
82 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 84 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
83 protected StorageService storageService; 85 protected StorageService storageService;
84 86
87 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
88 + protected KafkaConfigService kafkaConfigService;
89 +
85 private ApplicationId appId; 90 private ApplicationId appId;
86 91
87 @Activate 92 @Activate
...@@ -100,13 +105,11 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -100,13 +105,11 @@ public class EventSubscriptionManager implements EventSubscriptionService {
100 subscriptions = storageService 105 subscriptions = storageService
101 .<Type, List<EventSubscriber>>consistentMapBuilder() 106 .<Type, List<EventSubscriber>>consistentMapBuilder()
102 .withName(SUBSCRIBED_APPS) 107 .withName(SUBSCRIBED_APPS)
103 - .withSerializer(Serializer.using(KryoNamespaces.API, 108 + .withSerializer(Serializer
104 - EventSubscriber.class, 109 + .using(KryoNamespaces.API, EventSubscriber.class,
105 - OnosEvent.class, 110 + OnosEvent.class, OnosEvent.Type.class,
106 - OnosEvent.Type.class, 111 + DefaultEventSubscriber.class,
107 - DefaultEventSubscriber.class, 112 + EventSubscriberGroupId.class, UUID.class))
108 - EventSubscriberGroupId.class,
109 - UUID.class))
110 .build().asJavaMap(); 113 .build().asJavaMap();
111 114
112 log.info("Started"); 115 log.info("Started");
...@@ -118,15 +121,22 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -118,15 +121,22 @@ public class EventSubscriptionManager implements EventSubscriptionService {
118 } 121 }
119 122
120 @Override 123 @Override
121 - public EventSubscriberGroupId registerListener(String appName) { 124 + public RegistrationResponse registerListener(String appName) {
122 125
123 // TODO: Remove it once ONOS provides a mechanism for external apps 126 // TODO: Remove it once ONOS provides a mechanism for external apps
124 // to register with the core service. See Jira - 4409 127 // to register with the core service. See Jira - 4409
125 ApplicationId externalAppId = coreService.registerApplication(appName); 128 ApplicationId externalAppId = coreService.registerApplication(appName);
126 129
127 - return registeredApps.computeIfAbsent(externalAppId, 130 + EventSubscriberGroupId id =
128 - (key) -> new EventSubscriberGroupId(UUID 131 + registeredApps.computeIfAbsent(externalAppId,
129 - .randomUUID())); 132 + (key) -> new EventSubscriberGroupId(UUID
133 + .randomUUID()));
134 +
135 + RegistrationResponse response = new RegistrationResponse(id,
136 + kafkaConfigService.getConfigParams().getIpAddress(),
137 + kafkaConfigService.getConfigParams().getPort());
138 +
139 + return response;
130 } 140 }
131 141
132 @Override 142 @Override
...@@ -147,16 +157,11 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -147,16 +157,11 @@ public class EventSubscriptionManager implements EventSubscriptionService {
147 + "registered to make this request."); 157 + "registered to make this request.");
148 } 158 }
149 159
150 - if (!validGroupId(subscriber.subscriberGroupId(), subscriber.appName())) { 160 + if (!validGroupId(subscriber.subscriberGroupId(),
161 + subscriber.appName())) {
151 throw new InvalidGroupIdException("Incorrect group id in the request"); 162 throw new InvalidGroupIdException("Incorrect group id in the request");
152 } 163 }
153 164
154 - OnosEventListener onosListener = getListener(subscriber.eventType());
155 - checkNotNull(onosListener, "No listener for the supported event type - {}", subscriber.eventType());
156 -
157 - applyListenerAction(subscriber.eventType(), onosListener,
158 - ListenerAction.START);
159 -
160 // update internal state 165 // update internal state
161 List<EventSubscriber> subscriptionList = 166 List<EventSubscriber> subscriptionList =
162 subscriptions.get(subscriber.eventType()); 167 subscriptions.get(subscriber.eventType());
...@@ -233,12 +238,6 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -233,12 +238,6 @@ public class EventSubscriptionManager implements EventSubscriptionService {
233 * @param eventType ONOS event type 238 * @param eventType ONOS event type
234 * @return ONOS event listener 239 * @return ONOS event listener
235 */ 240 */
236 - private OnosEventListener getListener(Type eventType) {
237 - checkNotNull(eventType);
238 - ListenerFactory factory = ListenerFactory.getInstance();
239 - OnosEventListener onosListener = factory.getListener(eventType);
240 - return onosListener;
241 - }
242 241
243 /** 242 /**
244 * Checks if the group id is valid for this registered application. 243 * Checks if the group id is valid for this registered application.
...@@ -287,15 +286,6 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -287,15 +286,6 @@ public class EventSubscriptionManager implements EventSubscriptionService {
287 // stop the listener. 286 // stop the listener.
288 List<EventSubscriber> subscribers = 287 List<EventSubscriber> subscribers =
289 subscriptions.get(subscriber.eventType()); 288 subscriptions.get(subscriber.eventType());
290 - if (subscribers.size() == 1) {
291 - OnosEventListener onosListener =
292 - getListener(subscriber.eventType());
293 - checkNotNull(onosListener,
294 - "No listener for the supported event type - {}",
295 - subscriber.eventType());
296 - applyListenerAction(subscriber.eventType(), onosListener,
297 - ListenerAction.STOP);
298 - }
299 289
300 // update internal state. 290 // update internal state.
301 subscribers.remove(subscriber); 291 subscribers.remove(subscriber);
...@@ -308,7 +298,6 @@ public class EventSubscriptionManager implements EventSubscriptionService { ...@@ -308,7 +298,6 @@ public class EventSubscriptionManager implements EventSubscriptionService {
308 @Override 298 @Override
309 public List<EventSubscriber> getEventSubscribers(Type type) { 299 public List<EventSubscriber> getEventSubscribers(Type type) {
310 return subscriptions.getOrDefault(type, ImmutableList.of()); 300 return subscriptions.getOrDefault(type, ImmutableList.of());
311 -
312 } 301 }
313 302
314 /** 303 /**
......
...@@ -31,7 +31,7 @@ import java.util.concurrent.Executors; ...@@ -31,7 +31,7 @@ import java.util.concurrent.Executors;
31 import java.util.concurrent.ScheduledExecutorService; 31 import java.util.concurrent.ScheduledExecutorService;
32 import java.util.concurrent.TimeUnit; 32 import java.util.concurrent.TimeUnit;
33 33
34 -@Component(immediate = true) 34 +@Component(immediate = false)
35 public class KafkaStorageManager implements KafkaEventStorageService { 35 public class KafkaStorageManager implements KafkaEventStorageService {
36 36
37 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 37 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
......
1 +/**
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.kafkaintegration.kafka;
17 +
18 +import static com.google.common.base.Strings.isNullOrEmpty;
19 +import static org.onlab.util.Tools.get;
20 +
21 +import java.util.Dictionary;
22 +
23 +import org.apache.felix.scr.annotations.Activate;
24 +import org.apache.felix.scr.annotations.Component;
25 +import org.apache.felix.scr.annotations.Deactivate;
26 +import org.apache.felix.scr.annotations.Modified;
27 +import org.apache.felix.scr.annotations.Property;
28 +import org.apache.felix.scr.annotations.Reference;
29 +import org.apache.felix.scr.annotations.ReferenceCardinality;
30 +import org.apache.felix.scr.annotations.Service;
31 +import org.onosproject.cfg.ComponentConfigService;
32 +import org.onosproject.kafkaintegration.api.KafkaConfigService;
33 +import org.onosproject.kafkaintegration.api.KafkaProducerService;
34 +import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
35 +import org.osgi.service.component.ComponentContext;
36 +import org.slf4j.Logger;
37 +import org.slf4j.LoggerFactory;
38 +
39 +@Component(immediate = true)
40 +@Service
41 +public class KafkaConfigManager implements KafkaConfigService {
42 +
43 + private final Logger log = LoggerFactory.getLogger(getClass());
44 +
45 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
46 + protected ComponentConfigService componentConfigService;
47 +
48 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
49 + protected KafkaProducerService producer;
50 +
51 + public static final String BOOTSTRAP_SERVERS = "localhost:9092";
52 + private String kafkaServerIp =
53 + BOOTSTRAP_SERVERS.substring(0, BOOTSTRAP_SERVERS.indexOf(":"));
54 + private String kafkaServerPortNum =
55 + BOOTSTRAP_SERVERS.substring(BOOTSTRAP_SERVERS.indexOf(":") + 1,
56 + BOOTSTRAP_SERVERS.length());
57 +
58 + private static final int RETRIES = 1;
59 + private static final int MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 5;
60 + private static final int REQUEST_REQUIRED_ACKS = 1;
61 + private static final String KEY_SERIALIZER =
62 + "org.apache.kafka.common.serialization.StringSerializer";
63 + private static final String VALUE_SERIALIZER =
64 + "org.apache.kafka.common.serialization.ByteArraySerializer";
65 +
66 + @Property(name = "bootstrap.servers", value = BOOTSTRAP_SERVERS,
67 + label = "Default IP/Port pair to establish initial connection to Kafka cluster.")
68 + protected String bootstrapServers = BOOTSTRAP_SERVERS;
69 +
70 + @Property(name = "retries", intValue = RETRIES,
71 + label = "Number of times the producer can retry to send after first failure")
72 + protected int retries = RETRIES;
73 +
74 + @Property(name = "max.in.flight.requests.per.connection",
75 + intValue = MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
76 + label = "The maximum number of unacknowledged requests the client will send before blocking")
77 + protected int maxInFlightRequestsPerConnection =
78 + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION;
79 +
80 + @Property(name = "request.required.acks", intValue = 1,
81 + label = "Producer will get an acknowledgement after the leader has replicated the data")
82 + protected int requestRequiredAcks = REQUEST_REQUIRED_ACKS;
83 +
84 + @Property(name = "key.serializer", value = KEY_SERIALIZER,
85 + label = "Serializer class for key that implements the Serializer interface.")
86 + protected String keySerializer = KEY_SERIALIZER;
87 +
88 + @Property(name = "value.serializer", value = VALUE_SERIALIZER,
89 + label = "Serializer class for value that implements the Serializer interface.")
90 + protected String valueSerializer = VALUE_SERIALIZER;
91 +
92 + @Activate
93 + protected void activate(ComponentContext context) {
94 + componentConfigService.registerProperties(getClass());
95 + log.info("Started");
96 + }
97 +
98 + @Deactivate
99 + protected void deactivate() {
100 + componentConfigService.unregisterProperties(getClass(), false);
101 + log.info("Stopped");
102 + }
103 +
104 + @Modified
105 + private void modified(ComponentContext context) {
106 + if (context == null) {
107 + bootstrapServers = BOOTSTRAP_SERVERS;
108 + retries = RETRIES;
109 + maxInFlightRequestsPerConnection =
110 + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION;
111 + requestRequiredAcks = REQUEST_REQUIRED_ACKS;
112 + keySerializer = KEY_SERIALIZER;
113 + valueSerializer = VALUE_SERIALIZER;
114 + return;
115 + }
116 +
117 + Dictionary<?, ?> properties = context.getProperties();
118 +
119 + String newBootstrapServers;
120 + int newRetries;
121 + int newMaxInFlightRequestsPerConnection;
122 + int newRequestRequiredAcks;
123 + try {
124 + String s = get(properties, "bootstrap.servers");
125 + newBootstrapServers =
126 + isNullOrEmpty(s) ? bootstrapServers : s.trim();
127 +
128 + s = get(properties, "retries");
129 + newRetries =
130 + isNullOrEmpty(s) ? retries : Integer.parseInt(s.trim());
131 +
132 + s = get(properties, "max.in.flight.requests.per.connection");
133 + newMaxInFlightRequestsPerConnection =
134 + isNullOrEmpty(s) ? maxInFlightRequestsPerConnection
135 + : Integer.parseInt(s.trim());
136 +
137 + s = get(properties, "request.required.acks");
138 + newRequestRequiredAcks =
139 + isNullOrEmpty(s) ? requestRequiredAcks
140 + : Integer.parseInt(s.trim());
141 +
142 + } catch (NumberFormatException | ClassCastException e) {
143 + return;
144 + }
145 +
146 + if (configModified(newBootstrapServers, newRetries,
147 + newMaxInFlightRequestsPerConnection,
148 + newRequestRequiredAcks)) {
149 + bootstrapServers = newBootstrapServers;
150 + kafkaServerIp = bootstrapServers
151 + .substring(0, bootstrapServers.indexOf(":"));
152 + kafkaServerPortNum = bootstrapServers
153 + .substring(bootstrapServers.indexOf(":") + 1,
154 + bootstrapServers.length());
155 +
156 + retries = newRetries;
157 +
158 + maxInFlightRequestsPerConnection =
159 + newMaxInFlightRequestsPerConnection;
160 +
161 + requestRequiredAcks = newRequestRequiredAcks;
162 +
163 + producer.restart(KafkaServerConfig.builder()
164 + .ipAddress(kafkaServerIp).port(kafkaServerPortNum)
165 + .numOfRetries(retries)
166 + .maxInFlightRequestsPerConnection(maxInFlightRequestsPerConnection)
167 + .acksRequired(requestRequiredAcks)
168 + .keySerializer(keySerializer)
169 + .valueSerializer(valueSerializer).build());
170 +
171 + log.info("Kafka Server Config has been Modified - "
172 + + "bootstrapServers {}, retries {}, "
173 + + "maxInFlightRequestsPerConnection {}, "
174 + + "requestRequiredAcks {}", bootstrapServers, retries,
175 + maxInFlightRequestsPerConnection, requestRequiredAcks);
176 + } else {
177 + return;
178 + }
179 + }
180 +
181 + private boolean configModified(String newBootstrapServers, int newRetries,
182 + int newMaxInFlightRequestsPerConnection,
183 + int newRequestRequiredAcks) {
184 +
185 + return !newBootstrapServers.equals(bootstrapServers)
186 + || newRetries != retries
187 + || newMaxInFlightRequestsPerConnection != maxInFlightRequestsPerConnection
188 + || newRequestRequiredAcks != requestRequiredAcks;
189 +
190 + }
191 +
192 + @Override
193 + public KafkaServerConfig getConfigParams() {
194 + String ipAddr =
195 + bootstrapServers.substring(0, bootstrapServers.indexOf(":"));
196 + String port =
197 + bootstrapServers.substring(bootstrapServers.indexOf(":") + 1,
198 + bootstrapServers.length());
199 +
200 + return KafkaServerConfig.builder().ipAddress(ipAddr).port(port)
201 + .numOfRetries(retries)
202 + .maxInFlightRequestsPerConnection(maxInFlightRequestsPerConnection)
203 + .acksRequired(requestRequiredAcks).keySerializer(keySerializer)
204 + .valueSerializer(valueSerializer).build();
205 +
206 + }
207 +
208 +}
...@@ -16,50 +16,84 @@ ...@@ -16,50 +16,84 @@
16 16
17 package org.onosproject.kafkaintegration.kafka; 17 package org.onosproject.kafkaintegration.kafka;
18 18
19 +import java.util.Properties;
20 +import java.util.concurrent.Future;
21 +
22 +import org.apache.felix.scr.annotations.Activate;
23 +import org.apache.felix.scr.annotations.Component;
24 +import org.apache.felix.scr.annotations.Deactivate;
25 +import org.apache.felix.scr.annotations.Service;
19 import org.apache.kafka.clients.producer.KafkaProducer; 26 import org.apache.kafka.clients.producer.KafkaProducer;
20 import org.apache.kafka.clients.producer.ProducerRecord; 27 import org.apache.kafka.clients.producer.ProducerRecord;
21 import org.apache.kafka.clients.producer.RecordMetadata; 28 import org.apache.kafka.clients.producer.RecordMetadata;
29 +import org.onosproject.kafkaintegration.api.KafkaProducerService;
30 +import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
22 import org.slf4j.Logger; 31 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory; 32 import org.slf4j.LoggerFactory;
24 33
25 -import java.util.Properties;
26 -import java.util.concurrent.Future;
27 -
28 /** 34 /**
29 - * Implementation of Kafka Producer. 35 + * Implementation of a Kafka Producer.
30 */ 36 */
31 -public class Producer { 37 +@Component
38 +@Service
39 +public class Producer implements KafkaProducerService {
32 private KafkaProducer<String, byte[]> kafkaProducer = null; 40 private KafkaProducer<String, byte[]> kafkaProducer = null;
33 41
34 private final Logger log = LoggerFactory.getLogger(getClass()); 42 private final Logger log = LoggerFactory.getLogger(getClass());
35 43
36 - Producer(String bootstrapServers, int retries, int maxInFlightRequestsPerConnection, 44 + @Activate
37 - int requestRequiredAcks, String keySerializer, String valueSerializer) { 45 + protected void activate() {
46 + log.info("Started");
47 + }
48 +
49 + @Deactivate
50 + protected void deactivate() {
51 + log.info("Stopped");
52 + }
53 +
54 + @Override
55 + public void start(KafkaServerConfig config) {
56 +
57 + if (kafkaProducer != null) {
58 + log.info("Producer has already started");
59 + return;
60 + }
61 +
62 + String bootstrapServer =
63 + new StringBuilder().append(config.getIpAddress()).append(":")
64 + .append(config.getPort()).toString();
38 65
66 + // Set Server Properties
39 Properties prop = new Properties(); 67 Properties prop = new Properties();
40 - prop.put("bootstrap.servers", bootstrapServers); 68 + prop.put("bootstrap.servers", bootstrapServer);
41 - prop.put("retries", retries); 69 + prop.put("retries", config.getNumOfRetries());
42 - prop.put("max.in.flight.requests.per.connection", maxInFlightRequestsPerConnection); 70 + prop.put("max.in.flight.requests.per.connection",
43 - prop.put("request.required.acks", requestRequiredAcks); 71 + config.getMaxInFlightRequestsPerConnection());
44 - prop.put("key.serializer", keySerializer); 72 + prop.put("request.required.acks", config.getAcksRequired());
45 - prop.put("value.serializer", valueSerializer); 73 + prop.put("key.serializer", config.getKeySerializer());
74 + prop.put("value.serializer", config.getValueSerializer());
46 75
47 kafkaProducer = new KafkaProducer<>(prop); 76 kafkaProducer = new KafkaProducer<>(prop);
77 + log.info("Kafka Producer has started.");
48 } 78 }
49 79
50 - public void start() { 80 + @Override
51 - log.info("Started");
52 - }
53 -
54 public void stop() { 81 public void stop() {
55 if (kafkaProducer != null) { 82 if (kafkaProducer != null) {
56 kafkaProducer.close(); 83 kafkaProducer.close();
57 kafkaProducer = null; 84 kafkaProducer = null;
58 } 85 }
59 86
60 - log.info("Stopped"); 87 + log.info("Kafka Producer has Stopped");
88 + }
89 +
90 + @Override
91 + public void restart(KafkaServerConfig config) {
92 + stop();
93 + start(config);
61 } 94 }
62 95
96 + @Override
63 public Future<RecordMetadata> send(ProducerRecord<String, byte[]> record) { 97 public Future<RecordMetadata> send(ProducerRecord<String, byte[]> record) {
64 return kafkaProducer.send(record); 98 return kafkaProducer.send(record);
65 } 99 }
......
1 -/**
2 - * Copyright 2016 Open Networking Laboratory
3 - * Licensed under the Apache License, Version 2.0 (the "License");
4 - * you may not use this file except in compliance with the License.
5 - * You may obtain a copy of the License at
6 -
7 - * http://www.apache.org/licenses/LICENSE-2.0
8 -
9 - * Unless required by applicable law or agreed to in writing, software
10 - * distributed under the License is distributed on an "AS IS" BASIS,
11 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 - * See the License for the specific language governing permissions and
13 - * limitations under the License.
14 - */
15 -package org.onosproject.kafkaintegration.listener;
16 -
17 -import com.google.protobuf.GeneratedMessage;
18 -import org.onosproject.event.ListenerService;
19 -import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
20 -import org.onosproject.kafkaintegration.converter.ConversionFactory;
21 -import org.onosproject.kafkaintegration.converter.EventConverter;
22 -import org.onosproject.kafkaintegration.impl.KafkaPublisherManager;
23 -import org.onosproject.net.device.DeviceEvent;
24 -import org.onosproject.net.device.DeviceListener;
25 -import org.onosproject.net.device.DeviceService;
26 -import org.slf4j.Logger;
27 -import org.slf4j.LoggerFactory;
28 -
29 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
30 -
31 -/**
32 - * Listens for ONOS Device events.
33 - *
34 - */
35 -public final class DeviceEventsListener implements OnosEventListener {
36 -
37 - private final Logger log = LoggerFactory.getLogger(getClass());
38 -
39 - private boolean listenerRunning = false;
40 -
41 - private InnerListener listener = null;
42 - // Exists to defeat instantiation
43 - private DeviceEventsListener() {
44 - }
45 -
46 - private static class SingletonHolder {
47 - private static final DeviceEventsListener INSTANCE =
48 - new DeviceEventsListener();
49 - }
50 -
51 - /**
52 - * Returns a static reference to the Listener Factory.
53 - *
54 - * @return singleton object
55 - */
56 - public static DeviceEventsListener getInstance() {
57 - return SingletonHolder.INSTANCE;
58 - }
59 -
60 - @Override
61 - public void startListener(Type e, ListenerService<?, ?> service) {
62 - if (!listenerRunning) {
63 - listener = new InnerListener();
64 - DeviceService deviceService = (DeviceService) service;
65 - deviceService.addListener(listener);
66 - listenerRunning = true;
67 - }
68 - }
69 -
70 - private class InnerListener implements DeviceListener {
71 -
72 - @Override
73 - public void event(DeviceEvent arg0) {
74 -
75 - // Convert the event to protobuf format
76 - ConversionFactory conversionFactory =
77 - ConversionFactory.getInstance();
78 - EventConverter converter = conversionFactory.getConverter(DEVICE);
79 - GeneratedMessage message = converter.convertToProtoMessage(arg0);
80 -
81 - // Call Dispatcher and publish event
82 - KafkaPublisherManager.getInstance().publish(DEVICE, message);
83 - }
84 - }
85 -
86 - @Override
87 - public void stopListener(Type e, ListenerService<?, ?> service) {
88 - if (listenerRunning) {
89 - DeviceService deviceService = (DeviceService) service;
90 - deviceService.removeListener(listener);
91 - listener = null;
92 - listenerRunning = false;
93 - }
94 - }
95 -
96 -}
...\ No newline at end of file ...\ No newline at end of file
1 -/**
2 - * Copyright 2016 Open Networking Laboratory
3 - * Licensed under the Apache License, Version 2.0 (the "License");
4 - * you may not use this file except in compliance with the License.
5 - * You may obtain a copy of the License at
6 -
7 - * http://www.apache.org/licenses/LICENSE-2.0
8 -
9 - * Unless required by applicable law or agreed to in writing, software
10 - * distributed under the License is distributed on an "AS IS" BASIS,
11 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 - * See the License for the specific language governing permissions and
13 - * limitations under the License.
14 - */
15 -package org.onosproject.kafkaintegration.listener;
16 -
17 -import com.google.protobuf.GeneratedMessage;
18 -import org.onosproject.event.ListenerService;
19 -import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
20 -import org.onosproject.kafkaintegration.converter.ConversionFactory;
21 -import org.onosproject.kafkaintegration.converter.EventConverter;
22 -import org.onosproject.kafkaintegration.impl.KafkaPublisherManager;
23 -import org.onosproject.net.link.LinkEvent;
24 -import org.onosproject.net.link.LinkListener;
25 -import org.onosproject.net.link.LinkService;
26 -
27 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
28 -
29 -/**
30 - * Listens for ONOS Link Events.
31 - *
32 - */
33 -public final class LinkEventsListener implements OnosEventListener {
34 -
35 - private boolean listenerRunning = false;
36 -
37 - private InnerListener listener = null;
38 - // Exists to defeat instantiation
39 - private LinkEventsListener() {
40 - }
41 -
42 - private static class SingletonHolder {
43 - private static final LinkEventsListener INSTANCE =
44 - new LinkEventsListener();
45 - }
46 -
47 - /**
48 - * Returns a static reference to the Listener Factory.
49 - *
50 - * @return singleton object
51 - */
52 - public static LinkEventsListener getInstance() {
53 - return SingletonHolder.INSTANCE;
54 - }
55 -
56 - @Override
57 - public void startListener(Type e, ListenerService<?, ?> service) {
58 - if (!listenerRunning) {
59 - listener = new InnerListener();
60 - LinkService linkService = (LinkService) service;
61 - linkService.addListener(listener);
62 - listenerRunning = true;
63 - }
64 - }
65 -
66 - private class InnerListener implements LinkListener {
67 -
68 - @Override
69 - public void event(LinkEvent arg0) {
70 -
71 - // Convert the event to protobuf format
72 - ConversionFactory conversionFactory =
73 - ConversionFactory.getInstance();
74 - EventConverter converter = conversionFactory.getConverter(LINK);
75 - GeneratedMessage message = converter.convertToProtoMessage(arg0);
76 -
77 - // Call Dispatcher and publish event
78 - KafkaPublisherManager.getInstance().publish(LINK, message);
79 - }
80 - }
81 -
82 - @Override
83 - public void stopListener(Type e, ListenerService<?, ?> service) {
84 - if (listenerRunning) {
85 - LinkService linkService = (LinkService) service;
86 - linkService.removeListener(listener);
87 - listenerRunning = false;
88 - }
89 - }
90 -}
1 -/**
2 - * Copyright 2016 Open Networking Laboratory
3 - * Licensed under the Apache License, Version 2.0 (the "License");
4 - * you may not use this file except in compliance with the License.
5 - * You may obtain a copy of the License at
6 -
7 - * http://www.apache.org/licenses/LICENSE-2.0
8 -
9 - * Unless required by applicable law or agreed to in writing, software
10 - * distributed under the License is distributed on an "AS IS" BASIS,
11 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 - * See the License for the specific language governing permissions and
13 - * limitations under the License.
14 - */
15 -package org.onosproject.kafkaintegration.listener;
16 -
17 -import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
18 -
19 -import java.util.HashMap;
20 -import java.util.Map;
21 -
22 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
23 -import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
24 -
25 -/**
26 - * Returns the appropriate listener object based on the ONOS event type.
27 - *
28 - */
29 -public final class ListenerFactory {
30 -
31 - // Store listeners for all supported events
32 - private Map<Type, OnosEventListener> listeners =
33 - new HashMap<Type, OnosEventListener>() {
34 - {
35 - put(DEVICE, DeviceEventsListener.getInstance());
36 - put(LINK, LinkEventsListener.getInstance());
37 - }
38 - };
39 -
40 - // Exists to defeat instantiation
41 - private ListenerFactory() {
42 - }
43 -
44 - private static class SingletonHolder {
45 - private static final ListenerFactory INSTANCE = new ListenerFactory();
46 - }
47 -
48 - /**
49 - * Returns a static reference to the Listener Factory.
50 - *
51 - * @return singleton object
52 - */
53 - public static ListenerFactory getInstance() {
54 - return SingletonHolder.INSTANCE;
55 - }
56 -
57 - /**
58 - * Returns the listener object for the specified ONOS event type.
59 - *
60 - * @param event ONOS Event type
61 - * @return return listener object
62 - */
63 - public OnosEventListener getListener(Type event) {
64 - return listeners.get(event);
65 - }
66 -
67 -}
...@@ -200,6 +200,7 @@ ...@@ -200,6 +200,7 @@
200 </execution> 200 </execution>
201 <execution> 201 <execution>
202 <id>swagger</id> 202 <id>swagger</id>
203 + <phase>generate-resources</phase>
203 <goals> 204 <goals>
204 <goal>swagger</goal> 205 <goal>swagger</goal>
205 </goals> 206 </goals>
......
...@@ -18,12 +18,11 @@ import com.fasterxml.jackson.databind.ObjectMapper; ...@@ -18,12 +18,11 @@ import com.fasterxml.jackson.databind.ObjectMapper;
18 import com.fasterxml.jackson.databind.node.ObjectNode; 18 import com.fasterxml.jackson.databind.node.ObjectNode;
19 import org.onosproject.kafkaintegration.api.EventSubscriptionService; 19 import org.onosproject.kafkaintegration.api.EventSubscriptionService;
20 import org.onosproject.kafkaintegration.api.dto.EventSubscriber; 20 import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
21 -import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId; 21 +import org.onosproject.kafkaintegration.api.dto.RegistrationResponse;
22 import org.onosproject.rest.AbstractWebResource; 22 import org.onosproject.rest.AbstractWebResource;
23 import org.slf4j.Logger; 23 import org.slf4j.Logger;
24 import org.slf4j.LoggerFactory; 24 import org.slf4j.LoggerFactory;
25 25
26 -
27 import javax.ws.rs.Consumes; 26 import javax.ws.rs.Consumes;
28 import javax.ws.rs.DELETE; 27 import javax.ws.rs.DELETE;
29 import javax.ws.rs.POST; 28 import javax.ws.rs.POST;
...@@ -62,7 +61,7 @@ public class EventExporterWebResource extends AbstractWebResource { ...@@ -62,7 +61,7 @@ public class EventExporterWebResource extends AbstractWebResource {
62 * 61 *
63 * @param appName The application trying to register 62 * @param appName The application trying to register
64 * @return 200 OK with UUID string which should be used as Kafka Consumer 63 * @return 200 OK with UUID string which should be used as Kafka Consumer
65 - * Group Id 64 + * Group Id and Kafka Server, port information.
66 * @onos.rsModel KafkaRegistration 65 * @onos.rsModel KafkaRegistration
67 */ 66 */
68 @POST 67 @POST
...@@ -73,12 +72,16 @@ public class EventExporterWebResource extends AbstractWebResource { ...@@ -73,12 +72,16 @@ public class EventExporterWebResource extends AbstractWebResource {
73 72
74 EventSubscriptionService service = get(EventSubscriptionService.class); 73 EventSubscriptionService service = get(EventSubscriptionService.class);
75 74
76 - EventSubscriberGroupId groupId = service.registerListener(appName); 75 + RegistrationResponse response = service.registerListener(appName);
76 +
77 + ObjectNode result = mapper().createObjectNode();
78 + result.put("groupId", response.getGroupId().getId().toString());
79 + result.put("ipAddress", response.getIpAddress());
80 + result.put("port", response.getPort());
77 81
78 log.info("Registered app {}", appName); 82 log.info("Registered app {}", appName);
79 - // TODO: Should also return Kafka server information. 83 +
80 - // Will glue this in when we have the config and Kafka modules ready 84 + return ok(result.toString()).build();
81 - return ok(groupId.getId().toString()).build();
82 } 85 }
83 86
84 /** 87 /**
......