fix: create consumer if its not present on remote site

frappe/events_streaming/doctype/change_request/change_request.json

@@ -4,19 +4,13 @@
  "editable_grid": 1,
  "engine": "InnoDB",
  "field_order": [
-  "follower_node",
+  "event_consumer",
   "ref_doctype",
   "document_name",
   "requested_change",
   "request_status"
  ],
  "fields": [
-  {
-   "fieldname": "follower_node",
-   "fieldtype": "Link",
-   "label": "Consumer Node",
-   "options": "Node"
-  },
   {
    "fieldname": "ref_doctype",
    "fieldtype": "Data",
@@ -37,9 +31,15 @@
    "fieldtype": "Select",
    "label": "Request Status",
    "options": "\nApproved\nRejected\nResolve Conflict"
+  },
+  {
+   "fieldname": "event_consumer",
+   "fieldtype": "Link",
+   "label": "Event Consumer",
+   "options": "Event Consumer"
   }
  ],
- "modified": "2019-08-26 14:03:56.911041",
+ "modified": "2019-09-05 13:44:06.934893",
  "modified_by": "Administrator",
  "module": "Events Streaming",
  "name": "Change Request",

frappe/events_streaming/doctype/event_producer/event_producer.py

@@ -8,7 +8,6 @@ import json
 from frappe import _
 from frappe.model.document import Document
 from frappe.frappeclient import FrappeClient
-# from frappe.desk.linked_with import get_linked_doctype
 
 class EventProducer(Document):
 	def after_insert(self):
@@ -19,22 +18,24 @@ class EventProducer(Document):
 
 	def update_event_consumer(self):
 		producer_site = get_producer_site(self.producer_url)
-		event_consumer = producer_site.get_doc('Event Consumer', get_current_node())
-		if not event_consumer:
+		try:
+			event_consumer = producer_site.get_doc('Event Consumer', get_current_node())
+			event_consumer.subscribed_doctypes = []
+			for entry in self.subscribed_doctypes:
+				event_consumer.subscribed_doctypes.append({
+					'ref_doctype': entry.ref_doctype
+				})
+			event_consumer.event_subscriber = self.event_subscriber
+			producer_site.update(event_consumer)
+		except Exception:
 			consumer = frappe.new_doc('Event Consumer')
 			consumer.callback_url = get_current_node()
 			for entry in self.subscribed_doctypes:
 				consumer.append('subscribed_doctypes', {
 					'ref_doctype': entry.ref_doctype
 				})
-			producer_site.insert(consumer)	
-		else:
-			event_consumer.subscribed_doctypes = []
-			for entry in self.subscribed_doctypes:
-				event_consumer.subscribed_doctypes.append({
-					'ref_doctype': entry.ref_doctype
-				})
-			producer_site.update(event_consumer)
+			consumer.event_subscriber = self.event_subscriber
+			producer_site.insert(consumer)
 
 def get_current_node():
 	current_node = frappe.utils.get_url()

frappe/model/document.py

@@ -1295,7 +1295,7 @@ def make_update_log(doc, update_type):
 		frappe.db.commit()
 
 def check_doctype_has_consumers(doctype):
-	event_consumers = frappe.get_all(doctype = 'Event Consumer')
+	event_consumers = frappe.get_all('Event Consumer')
 	for event_consumer in event_consumers:
 		consumer = frappe.get_doc('Event Consumer', event_consumer.name)
 		for entry in consumer.subscribed_doctypes: