Madan Jampani
Committed by Madan Jampani

Added custom transport implementaion (for Catalyst Transport) for all copycat specific communication

Change-Id: I801d973b7c3412f6a8efcec77fe73fc480b2ce6e
...@@ -32,9 +32,10 @@ ...@@ -32,9 +32,10 @@
32 <description>ONOS Core Store subsystem</description> 32 <description>ONOS Core Store subsystem</description>
33 33
34 <modules> 34 <modules>
35 + <module>primitives</module>
35 <module>dist</module> 36 <module>dist</module>
36 <module>persistence</module> 37 <module>persistence</module>
37 - <module>serializers</module> 38 + <module>serializers</module>
38 </modules> 39 </modules>
39 40
40 <dependencies> 41 <dependencies>
......
1 +<?xml version="1.0" encoding="UTF-8"?>
2 +<!--
3 + ~ Copyright 2016 Open Networking Laboratory
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 + ~ http://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 +<project xmlns="http://maven.apache.org/POM/4.0.0"
18 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
19 + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
20 + <modelVersion>4.0.0</modelVersion>
21 +
22 + <parent>
23 + <groupId>org.onosproject</groupId>
24 + <artifactId>onos-core-store</artifactId>
25 + <version>1.5.0-SNAPSHOT</version>
26 + <relativePath>../pom.xml</relativePath>
27 + </parent>
28 +
29 + <artifactId>onos-core-primitives</artifactId>
30 + <packaging>bundle</packaging>
31 +
32 + <description>ONOS distributed state management primitives</description>
33 +
34 + <dependencies>
35 +
36 + <dependency>
37 + <groupId>org.onosproject</groupId>
38 + <artifactId>onos-api</artifactId>
39 + </dependency>
40 +
41 + <!-- for shaded atomix/copycat -->
42 + <dependency>
43 + <groupId>org.onosproject</groupId>
44 + <artifactId>onlab-thirdparty</artifactId>
45 + </dependency>
46 + </dependencies>
47 +</project>
1 +/*
2 + * Copyright 2016 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.store.primitives.impl;
17 +
18 +import static com.google.common.base.Preconditions.checkNotNull;
19 +
20 +import org.onosproject.store.cluster.messaging.MessagingService;
21 +
22 +import io.atomix.catalyst.transport.Client;
23 +import io.atomix.catalyst.transport.Server;
24 +import io.atomix.catalyst.transport.Transport;
25 +
26 +/**
27 + * Custom {@link Transport transport} for Copycat interactions
28 + * built on top of {@link MessagingService}.
29 + *
30 + * @see CopycatTransportServer
31 + * @see CopycatTransportClient
32 + */
33 +public class CopycatTransport implements Transport {
34 +
35 + /**
36 + * Transport Mode.
37 + */
38 + public enum Mode {
39 + /**
40 + * Signifies transport for client -> server interaction.
41 + */
42 + CLIENT,
43 +
44 + /**
45 + * Signified transport for server -> server interaction.
46 + */
47 + SERVER
48 + }
49 +
50 + private final Mode mode;
51 + private final String clusterName;
52 + private final MessagingService messagingService;
53 +
54 + public CopycatTransport(Mode mode, String clusterName, MessagingService messagingService) {
55 + this.mode = checkNotNull(mode);
56 + this.clusterName = checkNotNull(clusterName);
57 + this.messagingService = checkNotNull(messagingService);
58 + }
59 +
60 + @Override
61 + public Client client() {
62 + return new CopycatTransportClient(clusterName,
63 + messagingService,
64 + mode);
65 + }
66 +
67 + @Override
68 + public Server server() {
69 + return new CopycatTransportServer(clusterName,
70 + messagingService);
71 + }
72 +}
1 +/*
2 + * Copyright 2016 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.store.primitives.impl;
17 +
18 +import static com.google.common.base.Preconditions.checkNotNull;
19 +
20 +import java.util.Set;
21 +import java.util.concurrent.CompletableFuture;
22 +import org.apache.commons.lang.math.RandomUtils;
23 +import org.onosproject.store.cluster.messaging.MessagingService;
24 +
25 +import com.google.common.collect.Sets;
26 +
27 +import io.atomix.catalyst.transport.Address;
28 +import io.atomix.catalyst.transport.Client;
29 +import io.atomix.catalyst.transport.Connection;
30 +import io.atomix.catalyst.util.concurrent.ThreadContext;
31 +
32 +/**
33 + * {@link Client} implementation for {@link CopycatTransport}.
34 + */
35 +public class CopycatTransportClient implements Client {
36 +
37 + private final String clusterName;
38 + private final MessagingService messagingService;
39 + private final CopycatTransport.Mode mode;
40 + private final ThreadContext context;
41 + private final Set<CopycatTransportConnection> connections = Sets.newConcurrentHashSet();
42 +
43 + CopycatTransportClient(String clusterName, MessagingService messagingService, CopycatTransport.Mode mode) {
44 + this.clusterName = checkNotNull(clusterName);
45 + this.messagingService = checkNotNull(messagingService);
46 + this.mode = checkNotNull(mode);
47 + this.context = ThreadContext.currentContextOrThrow();
48 + }
49 +
50 + @Override
51 + public CompletableFuture<Connection> connect(Address remoteAddress) {
52 + CopycatTransportConnection connection = new CopycatTransportConnection(
53 + nextConnectionId(),
54 + CopycatTransport.Mode.CLIENT,
55 + clusterName,
56 + remoteAddress,
57 + messagingService,
58 + context);
59 + if (mode == CopycatTransport.Mode.CLIENT) {
60 + connection.setBidirectional();
61 + }
62 + connections.add(connection);
63 + return CompletableFuture.supplyAsync(() -> connection, context.executor());
64 + }
65 +
66 + @Override
67 + public CompletableFuture<Void> close() {
68 + return CompletableFuture.allOf(connections.stream().map(Connection::close).toArray(CompletableFuture[]::new));
69 + }
70 +
71 + private long nextConnectionId() {
72 + return RandomUtils.nextLong();
73 + }
74 + }
1 +/*
2 + * Copyright 2016 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.store.primitives.impl;
17 +
18 +import static com.google.common.base.Preconditions.checkNotNull;
19 +
20 +import java.io.ByteArrayInputStream;
21 +import java.io.DataInputStream;
22 +import java.io.IOException;
23 +import java.net.InetAddress;
24 +import java.net.InetSocketAddress;
25 +import java.util.Map;
26 +import java.util.concurrent.CompletableFuture;
27 +import java.util.concurrent.atomic.AtomicBoolean;
28 +import java.util.function.Consumer;
29 +
30 +import org.apache.commons.io.IOUtils;
31 +import org.onlab.util.Tools;
32 +import org.onosproject.store.cluster.messaging.MessagingService;
33 +
34 +import com.google.common.collect.Maps;
35 +
36 +import io.atomix.catalyst.transport.Address;
37 +import io.atomix.catalyst.transport.Connection;
38 +import io.atomix.catalyst.transport.Server;
39 +import io.atomix.catalyst.util.concurrent.SingleThreadContext;
40 +import io.atomix.catalyst.util.concurrent.ThreadContext;
41 +
42 +/**
43 + * {@link Server} implementation for {@link CopycatTransport}.
44 + */
45 +public class CopycatTransportServer implements Server {
46 +
47 + private final AtomicBoolean listening = new AtomicBoolean(false);
48 + private CompletableFuture<Void> listenFuture;
49 + private final String clusterName;
50 + private final MessagingService messagingService;
51 + private final String messageSubject;
52 + private final Map<Long, CopycatTransportConnection> connections = Maps.newConcurrentMap();
53 +
54 + CopycatTransportServer(String clusterName, MessagingService messagingService) {
55 + this.clusterName = checkNotNull(clusterName);
56 + this.messagingService = checkNotNull(messagingService);
57 + this.messageSubject = String.format("onos-copycat-%s", clusterName);
58 + }
59 +
60 + @Override
61 + public CompletableFuture<Void> listen(Address address, Consumer<Connection> listener) {
62 + if (listening.get()) {
63 + return CompletableFuture.completedFuture(null);
64 + }
65 + ThreadContext context = ThreadContext.currentContextOrThrow();
66 + synchronized (this) {
67 + if (listenFuture == null) {
68 + listenFuture = new CompletableFuture<>();
69 + listen(address, listener, context);
70 + }
71 + }
72 + return listenFuture;
73 + }
74 +
75 + public void listen(Address address, Consumer<Connection> listener, ThreadContext context) {
76 + messagingService.registerHandler(messageSubject, (sender, payload) -> {
77 + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(payload))) {
78 + long connectionId = input.readLong();
79 + InetAddress senderHost = InetAddress.getByAddress(sender.host().toOctets());
80 + int senderPort = sender.port();
81 + Address senderAddress = new Address(new InetSocketAddress(senderHost, senderPort));
82 + AtomicBoolean newConnection = new AtomicBoolean(false);
83 + CopycatTransportConnection connection = connections.computeIfAbsent(connectionId, k -> {
84 + newConnection.set(true);
85 + return new CopycatTransportConnection(connectionId,
86 + CopycatTransport.Mode.SERVER,
87 + clusterName,
88 + senderAddress,
89 + messagingService,
90 + getOrCreateContext(context));
91 + });
92 + byte[] request = IOUtils.toByteArray(input);
93 + return CompletableFuture.supplyAsync(
94 + () -> {
95 + if (newConnection.get()) {
96 + listener.accept(connection);
97 + }
98 + return connection;
99 + }, context.executor()).thenCompose(c -> c.handle(request));
100 + } catch (IOException e) {
101 + return Tools.exceptionalFuture(e);
102 + }
103 + });
104 + listening.set(true);
105 + context.execute(() -> {
106 + listenFuture.complete(null);
107 + });
108 + }
109 +
110 + @Override
111 + public CompletableFuture<Void> close() {
112 + messagingService.unregisterHandler(messageSubject);
113 + return CompletableFuture.completedFuture(null);
114 + }
115 +
116 + /**
117 + * Returns the current execution context or creates one.
118 + */
119 + private ThreadContext getOrCreateContext(ThreadContext parentContext) {
120 + ThreadContext context = ThreadContext.currentContext();
121 + if (context != null) {
122 + return context;
123 + }
124 + return new SingleThreadContext("copycat-transport-server-" + clusterName, parentContext.serializer().clone());
125 + }
126 +}
1 +/*
2 + * Copyright 2016 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 +
17 +/**
18 + * Implementation for distributed state management primitives.
19 + */
20 +package org.onosproject.store.primitives.impl;
...@@ -77,6 +77,8 @@ ...@@ -77,6 +77,8 @@
77 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> 77 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
78 <onos-build-conf.version>1.1</onos-build-conf.version> 78 <onos-build-conf.version>1.1</onos-build-conf.version>
79 <netty4.version>4.0.33.Final</netty4.version> 79 <netty4.version>4.0.33.Final</netty4.version>
80 + <!-- TODO: replace with final release version when it is out -->
81 + <atomix.version>0.1.0-beta4</atomix.version>
80 <copycat.version>0.5.1.onos</copycat.version> 82 <copycat.version>0.5.1.onos</copycat.version>
81 <openflowj.version>0.9.1.onos</openflowj.version> 83 <openflowj.version>0.9.1.onos</openflowj.version>
82 <onos-maven-plugin.version>1.8-SNAPSHOT</onos-maven-plugin.version> 84 <onos-maven-plugin.version>1.8-SNAPSHOT</onos-maven-plugin.version>
...@@ -87,6 +89,7 @@ ...@@ -87,6 +89,7 @@
87 <codehaus.jackson.version>1.9.13</codehaus.jackson.version> 89 <codehaus.jackson.version>1.9.13</codehaus.jackson.version>
88 <slf4j.version>1.7.6</slf4j.version> 90 <slf4j.version>1.7.6</slf4j.version>
89 <guava.version>19.0</guava.version> 91 <guava.version>19.0</guava.version>
92 + <commons.io.version>2.4</commons.io.version>
90 <!-- TODO argLine was originally added maven-surfire-plugin configuration 93 <!-- TODO argLine was originally added maven-surfire-plugin configuration
91 to fix locale errors for non-US developers. However, it breaks 94 to fix locale errors for non-US developers. However, it breaks
92 SonarQube's test coverage, so moving here for now. --> 95 SonarQube's test coverage, so moving here for now. -->
...@@ -801,4 +804,4 @@ ...@@ -801,4 +804,4 @@
801 </plugin> 804 </plugin>
802 </plugins> 805 </plugins>
803 </reporting> 806 </reporting>
804 -</project>
...\ No newline at end of file ...\ No newline at end of file
807 +</project>
......
...@@ -38,6 +38,18 @@ ...@@ -38,6 +38,18 @@
38 </dependency> 38 </dependency>
39 39
40 <dependency> 40 <dependency>
41 + <groupId>commons-io</groupId>
42 + <artifactId>commons-io</artifactId>
43 + <version>${commons.io.version}</version>
44 + </dependency>
45 +
46 + <dependency>
47 + <groupId>io.atomix</groupId>
48 + <artifactId>atomix</artifactId>
49 + <version>${atomix.version}</version>
50 + </dependency>
51 +
52 + <dependency>
41 <!-- FIXME once fixes get merged to upstream --> 53 <!-- FIXME once fixes get merged to upstream -->
42 <groupId>org.onosproject</groupId> 54 <groupId>org.onosproject</groupId>
43 <artifactId>copycat-api</artifactId> 55 <artifactId>copycat-api</artifactId>
...@@ -57,6 +69,7 @@ ...@@ -57,6 +69,7 @@
57 <plugin> 69 <plugin>
58 <groupId>org.apache.maven.plugins</groupId> 70 <groupId>org.apache.maven.plugins</groupId>
59 <artifactId>maven-shade-plugin</artifactId> 71 <artifactId>maven-shade-plugin</artifactId>
72 + <version>2.4.1</version>
60 <configuration> 73 <configuration>
61 <createSourcesJar>true</createSourcesJar> 74 <createSourcesJar>true</createSourcesJar>
62 75
...@@ -81,13 +94,35 @@ ...@@ -81,13 +94,35 @@
81 </filter> 94 </filter>
82 95
83 <filter> 96 <filter>
97 + <artifact>commons-io:commons-io</artifact>
98 + <includes>
99 + <include>org/apache/commons/io/**</include>
100 + </includes>
101 + </filter>
102 +
103 + <filter>
84 <artifact>org.onosproject:copycat*</artifact> 104 <artifact>org.onosproject:copycat*</artifact>
85 <includes> 105 <includes>
86 <include>**</include> 106 <include>**</include>
87 </includes> 107 </includes>
88 </filter> 108 </filter>
89 109
110 + <filter>
111 + <artifact>io.atomix:atomix-all</artifact>
112 + <includes>
113 + <include>**</include>
114 + </includes>
115 + </filter>
116 +
90 </filters> 117 </filters>
118 + <transformers>
119 + <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
120 + <resource>META-INF/services/io.atomix.catalyst.serializer.CatalystSerializable</resource>
121 + </transformer>
122 + <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
123 + <resource>META-INF/services/io.atomix.resource.Resource</resource>
124 + </transformer>
125 + </transformers>
91 </configuration> 126 </configuration>
92 <executions> 127 <executions>
93 <execution> 128 <execution>
...@@ -104,8 +139,11 @@ ...@@ -104,8 +139,11 @@
104 <configuration> 139 <configuration>
105 <instructions> 140 <instructions>
106 <Export-Package> 141 <Export-Package>
107 - com.googlecode.concurrenttrees.*;net.kuujo.copycat.* 142 + com.googlecode.concurrenttrees.*;net.kuujo.copycat.*;io.atomix.*
108 </Export-Package> 143 </Export-Package>
144 + <Import-Package>
145 + !sun.nio.ch,!sun.misc,*
146 + </Import-Package>
109 </instructions> 147 </instructions>
110 </configuration> 148 </configuration>
111 </plugin> 149 </plugin>
......