Repository: iluwatar/java-design-patterns
Issue: Publish-Subscribe Channel (EIP)
Labels: epic: pattern


Issue comments:
This pattern is similar in intent to Observer pattern. 

## Observer
- Is used for same process events
- Subscriber is aware of the publisher's existence
- It is one (pub) to many (subs)
- Publisher fires the on change events
- Events are fired in sync
- Implementation: `Observable` and `Observer` classes in Java

## Pub-Sub
- Typically used when there are two different processes
- Optionally there is topic based or content based filter
- There is a middle layer (queuing layer involved)
- Publishers and subscribers don't know about existence of each other. They just know about the existence of a channel (queue).

A friend of mine is very pleased with this pattern. So it's about time to present him an example in Java.

I'll offer some testable code these days.

What is going on here?

On Mon, Oct 5, 2015 at 9:12 AM, Michael Dähnert notifications@github.com
wrote:

> A friend of mine is very pleased with this pattern. So it's about time to
> present him an example in Java.
> 
> I'll offer some testable code these days.
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/iluwatar/java-design-patterns/issues/161#issuecomment-145585246
> .

I'm working on an implementation of this pattern. If you like, please alter the ticket labels.

Please note https://github.com/iluwatar/java-design-patterns/wiki/10.-Technology-selections

So you would have to implement the pattern using Apache Camel and then document how the framework implements this.

Let me know if you still want to take this issue and I will update the labels.

Thanks for the information. Yeah I'll do it. Seems to be much smarter to implement it with Apache Camel then on my own.

Ok I will alter the labels.

https://docs.microsoft.com/en-us/azure/architecture/patterns/publisher-subscriber

Pull request review comments:
You could have pluralized Subscriber. It sounds more natural.


Patch that resolved the issue:
diff --git a/pom.xml b/pom.xml
index 3064f3c0dd8f..e01dc10018b6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -80,6 +80,7 @@
 		<module>message-channel</module>
 		<module>fluentinterface</module>
 		<module>reactor</module>
+		<module>publish-subscribe</module>
 	</modules>
 
 	<dependencyManagement>
diff --git a/publish-subscribe/.gitignore b/publish-subscribe/.gitignore
new file mode 100644
index 000000000000..b83d22266ac8
--- /dev/null
+++ b/publish-subscribe/.gitignore
@@ -0,0 +1,1 @@
+/target/
diff --git a/publish-subscribe/etc/publish-subscribe.ucls b/publish-subscribe/etc/publish-subscribe.ucls
new file mode 100644
index 000000000000..1b121506eed8
--- /dev/null
+++ b/publish-subscribe/etc/publish-subscribe.ucls
@@ -0,0 +1,218 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<class-diagram version="1.1.8" icons="true" always-add-relationships="false" generalizations="true" realizations="true" 
+  associations="true" dependencies="false" nesting-relationships="true">  
+  <class id="1" language="java" name="org.apache.camel.builder.RouteBuilder" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="269" y="526"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <interface id="2" language="java" name="org.apache.camel.RoutesBuilder" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="270" y="406"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </interface>  
+  <class id="3" language="java" name="org.apache.camel.model.RouteDefinition" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="574" y="421"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <class id="4" language="java" name="org.apache.camel.model.RoutesDefinition" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="574" y="526"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <class id="5" language="java" name="org.apache.camel.model.MulticastDefinition" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="574" y="68"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <class id="6" language="java" name="org.apache.camel.model.OutputDefinition" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="576" y="164"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <class id="7" language="java" name="org.apache.camel.model.ProcessorDefinition" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="575" y="284"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <interface id="8" language="java" name="org.apache.camel.CamelContext" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_LEFT">    
+    <position height="-1" width="-1" x="1086" y="346"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </interface>  
+  <interface id="9" language="java" name="org.apache.camel.model.ModelCamelContext" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_LEFT">    
+    <position height="-1" width="-1" x="1086" y="526"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </interface>  
+  <class id="10" language="java" name="org.apache.camel.impl.DefaultCamelContext" project="message-channel" 
+    file="D:/Users/mdaehner/.m2/repository/org/apache/camel/camel-core/2.15.3/camel-core-2.15.3.jar" binary="true" 
+    corner="BOTTOM_RIGHT">    
+    <position height="-1" width="-1" x="885" y="421"/>    
+    <display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+      sort-features="false" accessors="true" visibility="true">      
+      <attributes public="false" package="false" protected="false" private="false" static="true"/>      
+      <operations public="false" package="false" protected="false" private="false" static="false"/>    
+    </display>  
+  </class>  
+  <realization id="11">    
+    <end type="SOURCE" refId="10"/>    
+    <end type="TARGET" refId="9"/>  
+  </realization>  
+  <generalization id="12">    
+    <end type="SOURCE" refId="6"/>    
+    <end type="TARGET" refId="7"/>  
+  </generalization>  
+  <association id="13">    
+    <end type="SOURCE" refId="6" navigable="false">      
+      <attribute id="14" name="outputs">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="15" minimum="0" maximum="2147483647">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="7" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <realization id="16">    
+    <end type="SOURCE" refId="1"/>    
+    <end type="TARGET" refId="2"/>  
+  </realization>  
+  <generalization id="17">    
+    <end type="SOURCE" refId="3"/>    
+    <end type="TARGET" refId="7"/>  
+  </generalization>  
+  <association id="18">    
+    <end type="SOURCE" refId="4" navigable="false">      
+      <attribute id="19" name="camelContext">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="20" minimum="0" maximum="1">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="9" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <association id="21">    
+    <end type="SOURCE" refId="7" navigable="false">      
+      <attribute id="22" name="parent">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="23" minimum="0" maximum="1">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="7" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <association id="24">    
+    <end type="SOURCE" refId="1" navigable="false">      
+      <attribute id="25" name="routeCollection">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="26" minimum="0" maximum="1">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="4" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <association id="27">    
+    <end type="SOURCE" refId="10" navigable="false">      
+      <attribute id="28" name="routeDefinitions">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="29" minimum="0" maximum="2147483647">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="3" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <association id="30">    
+    <end type="SOURCE" refId="3" navigable="false">      
+      <attribute id="31" name="outputs">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="32" minimum="0" maximum="2147483647">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="7" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <generalization id="33">    
+    <end type="SOURCE" refId="9"/>    
+    <end type="TARGET" refId="8"/>  
+  </generalization>  
+  <association id="34">    
+    <end type="SOURCE" refId="4" navigable="false">      
+      <attribute id="35" name="routes">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </attribute>      
+      <multiplicity id="36" minimum="0" maximum="2147483647">        
+        <position height="0" width="0" x="0" y="0"/>      
+      </multiplicity>    
+    </end>    
+    <end type="TARGET" refId="3" navigable="true"/>    
+    <display labels="true" multiplicity="true"/>  
+  </association>  
+  <generalization id="37">    
+    <end type="SOURCE" refId="5"/>    
+    <end type="TARGET" refId="6"/>  
+  </generalization>  
+  <classifier-display autosize="true" stereotype="true" package="true" initial-value="false" signature="false" 
+    sort-features="false" accessors="true" visibility="true">    
+    <attributes public="false" package="false" protected="false" private="false" static="true"/>    
+    <operations public="false" package="false" protected="false" private="false" static="false"/>  
+  </classifier-display>  
+  <association-display labels="true" multiplicity="true"/>
+</class-diagram>
\ No newline at end of file
diff --git a/publish-subscribe/index.md b/publish-subscribe/index.md
new file mode 100644
index 000000000000..efd40f02169b
--- /dev/null
+++ b/publish-subscribe/index.md
@@ -0,0 +1,17 @@
+---
+layout: pattern
+title: Publish Subscribe
+folder: publish-subscribe
+permalink: /patterns/publish-subscribe/
+categories: Integration
+tags: Java
+---
+
+**Intent:** When applications communicate using a messaging system they do it by using logical addresses
+of the system, so called Publish Subscribe Channel. The publisher broadcasts a message to all registered Subscriber.
+
+![alt text](./etc/publish-subscribe.png "Publish Subscribe Channel")
+
+**Applicability:** Use the Publish Subscribe Channel pattern when
+
+* two or more applications need to communicate using a messaging system for broadcasts.
diff --git a/publish-subscribe/pom.xml b/publish-subscribe/pom.xml
new file mode 100644
index 000000000000..04b76690afb0
--- /dev/null
+++ b/publish-subscribe/pom.xml
@@ -0,0 +1,24 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>com.iluwatar</groupId>
+		<artifactId>java-design-patterns</artifactId>
+		<version>1.7.0</version>
+	</parent>
+	<artifactId>publish-subscribe</artifactId>
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.camel</groupId>
+			<artifactId>camel-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.camel</groupId>
+			<artifactId>camel-stream</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+		</dependency>
+	</dependencies>
+</project>
\ No newline at end of file
diff --git a/publish-subscribe/src/main/java/com/iluwatar/publish/subscribe/App.java b/publish-subscribe/src/main/java/com/iluwatar/publish/subscribe/App.java
new file mode 100644
index 000000000000..1f433f5171a8
--- /dev/null
+++ b/publish-subscribe/src/main/java/com/iluwatar/publish/subscribe/App.java
@@ -0,0 +1,55 @@
+package com.iluwatar.publish.subscribe;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+
+/**
+ * 
+ * When applications communicate with each other using a messaging system
+ * they first need to establish a communication channel that will carry the
+ * data. Message Channel decouples Message producers (publisher) and consumers (subscriber).
+ * <p>
+ * The sending application doesn't necessarily know what particular applications
+ * will end up retrieving it, but it can be assured that the application that
+ * retrieves the information is interested in that information. This is because
+ * the messaging system has different Message Channels for different types of
+ * information the applications want to communicate. When an application sends
+ * information, it doesn't randomly add the information to any channel available;
+ * it adds it to a channel whose specific purpose is to communicate that sort of
+ * information. Likewise, an application that wants to receive particular information
+ * doesn't pull info off some random channel; it selects what channel to get information
+ * from based on what type of information it wants.
+ * <p>
+ * In this example we use Apache Camel to establish different Message Channels. The first
+ * one reads from standard input and delivers messages to Direct endpoints (Publish; Broadcast). The other Message  
+ * Channels are established from the Direct component to different Endpoints (Subscriber). No actual messages are sent, 
+ * only the established routes are printed to standard output.
+ * 
+ */
+public class App {
+
+	/**
+	 * Program entry point
+	 * 
+	 * @param args command line args
+	 * @throws Exception
+	 */
+	public static void main(String[] args) throws Exception {
+		CamelContext context = new DefaultCamelContext();
+		
+		context.addRoutes(new RouteBuilder() {
+
+			@Override
+			public void configure() throws Exception {
+				from("stream:in")
+				.multicast()
+				.to("direct:greetings1", "direct:greetings2", "direct:greetings3");
+			}
+		});
+		
+		context.start();
+		context.getRoutes().stream().forEach((r) -> System.out.println(r));
+		context.stop();
+	}
+}


Test patch:
diff --git a/publish-subscribe/src/test/java/com/iluwatar/publish/subscribe/AppTest.java b/publish-subscribe/src/test/java/com/iluwatar/publish/subscribe/AppTest.java
new file mode 100644
index 000000000000..743e4200d161
--- /dev/null
+++ b/publish-subscribe/src/test/java/com/iluwatar/publish/subscribe/AppTest.java
@@ -0,0 +1,17 @@
+package com.iluwatar.publish.subscribe;
+
+import org.junit.Test;
+
+/**
+ * 
+ * Application test
+ *
+ */
+public class AppTest {
+
+	@Test
+	public void test() throws Exception {
+		String[] args = {};
+		App.main(args);
+	}
+}
\ No newline at end of file