Merge pull request #81 from zenaton/feature/scheduling

Add scheduling

Gemfile.lock

@@ -13,7 +13,7 @@ GEM
       i18n (>= 0.7, < 2)
       minitest (~> 5.1)
       tzinfo (~> 1.1)
-    addressable (2.5.2)
+    addressable (2.6.0)
       public_suffix (>= 2.0.2, < 4.0)
     appraisal (2.2.0)
       bundler
@@ -26,7 +26,7 @@ GEM
       safe_yaml (~> 1.0.0)
     diff-lcs (1.3)
     docile (1.3.1)
-    hashdiff (0.3.7)
+    hashdiff (1.0.0)
     i18n (1.6.0)
       concurrent-ruby (~> 1.0)
     jaro_winkler (1.5.1)
@@ -40,7 +40,7 @@ GEM
     pry (0.11.3)
       coderay (~> 1.1.0)
       method_source (~> 0.9.0)
-    public_suffix (3.0.2)
+    public_suffix (3.1.1)
     rainbow (3.0.0)
     rake (10.5.0)
     rspec (3.7.0)
@@ -69,7 +69,7 @@ GEM
     rubocop-rspec (1.27.0)
       rubocop (>= 0.56.0)
     ruby-progressbar (1.9.0)
-    safe_yaml (1.0.4)
+    safe_yaml (1.0.5)
     simplecov (0.16.1)
       docile (~> 1.1)
       json (>= 1.8, < 3)
@@ -84,10 +84,10 @@ GEM
       tzinfo (>= 1.0.0)
     unicode-display_width (1.4.0)
     vcr (4.0.0)
-    webmock (3.4.2)
+    webmock (3.6.2)
       addressable (>= 2.3.6)
       crack (>= 0.3.2)
-      hashdiff
+      hashdiff (>= 0.4.0, < 2.0.0)
 
 PLATFORMS
   ruby

lib/zenaton/client.rb

@@ -2,6 +2,7 @@
 
 require 'securerandom'
 require 'singleton'
+require 'zenaton/services/graphql'
 require 'zenaton/services/http'
 require 'zenaton/services/properties'
 require 'zenaton/services/serializer'
@@ -13,6 +14,7 @@ class Client
     include Singleton
 
     ZENATON_API_URL = 'https://api.zenaton.com/v1' # Zenaton api url
+    ZENATON_GATEWAY_URL = 'https://gateway.zenaton.com/api' # Gateway url
     ZENATON_WORKER_URL = 'http://localhost' # Default worker url
     DEFAULT_WORKER_PORT = 4001 # Default worker port
     WORKER_API_VERSION = 'v_newton' # Default worker api version
@@ -61,6 +63,7 @@ def self.init(app_id, api_token, app_env)
     # @private
     def initialize
       @http = Services::Http.new
+      @graphql = Services::GraphQL.new(http: @http)
       @serializer = Services::Serializer.new
       @properties = Services::Properties.new
     end
@@ -81,6 +84,17 @@ def worker_url(resource = '', params = {})
       end
     end
 
+    def gateway_url
+      ENV['ZENATON_GATEWAY_URL'] || ZENATON_GATEWAY_URL
+    end
+
+    def gateway_headers
+      {
+        'app-id' => @app_id,
+        'api-token' => @api_token
+      }
+    end
+
     # Gets the url for zenaton api
     # @param resource [String] the endpoint for the api
     # @param params [Hash|String] query params to be url encoded
@@ -125,6 +139,55 @@ def start_workflow(flow)
       )
     end
 
+    def start_scheduled_task(task, cron)
+      res = @graphql.request(
+        gateway_url,
+        Services::GraphQL::CREATE_TASK_SCHEDULE,
+        create_task_schedule_input(task, cron),
+        gateway_headers
+      )
+      res && res['createTaskSchedule']
+    end
+
+    def start_scheduled_workflow(flow, cron)
+      res = @graphql.request(
+        gateway_url,
+        Services::GraphQL::CREATE_WORKFLOW_SCHEDULE,
+        create_workflow_schedule_input(flow, cron),
+        gateway_headers
+      )
+      res && res['createWorkflowSchedule']
+    end
+
+    # rubocop:disable Metrics/MethodLength
+    def create_workflow_schedule_input(flow, cron)
+      {
+        'createWorkflowScheduleInput' => {
+          'intentId' => SecureRandom.uuid,
+          'environmentName' => @app_env,
+          'cron' => cron,
+          'workflowName' => class_name(flow),
+          'canonicalName' => canonical_name(flow) || class_name(flow),
+          'programmingLanguage' => PROG.upcase,
+          'properties' => @serializer.encode(@properties.from(flow))
+        }
+      }
+    end
+    # rubocop:enable Metrics/MethodLength
+
+    def create_task_schedule_input(task, cron)
+      {
+        'createTaskScheduleInput' => {
+          'intentId' => SecureRandom.uuid,
+          'environmentName' => @app_env,
+          'cron' => cron,
+          'taskName' => class_name(task),
+          'programmingLanguage' => PROG.upcase,
+          'properties' => @serializer.encode(@properties.from(task))
+        }
+      }
+    end
+
     # Stops a workflow
     # @param workflow_name [String] the class name of the workflow
     # @param custom_id [String] the custom ID of the workflow (if any)

lib/zenaton/engine.rb

@@ -24,6 +24,22 @@ def initialize
       @processor = nil
     end
 
+    # Executes scheduling jobs synchronously
+    # @param jobs [Array<Zenaton::Interfaces::Job>]
+    # @param cron String
+    # @return [Array<String>, nil] the results if executed locally, or nil
+    def schedule(jobs, cron)
+      jobs.map(&method(:check_argument))
+      jobs.map do |job|
+        if job.is_a? Interfaces::Workflow
+          @client.start_scheduled_workflow(job, cron)
+        else
+          @client.start_scheduled_task(job, cron)
+        end
+      end
+      nil
+    end
+
     # Executes jobs synchronously
     # @param jobs [Array<Zenaton::Interfaces::Job>]
     # @return [Array<String>, nil] the results if executed locally, or nil

lib/zenaton/services/graphql.rb

@@ -0,0 +1,71 @@
+# frozen_string_literal: true
+
+require 'zenaton/exceptions'
+
+module Zenaton
+  # Collection of utility classes for the Zenaton library
+  module Services
+    # Service that:
+    # - handles graphql calls
+    # - translates exceptions into Zenaton specific ones
+    class GraphQL
+      CREATE_WORKFLOW_SCHEDULE = <<-'GRAPHQL'
+        mutation ($createWorkflowScheduleInput: CreateWorkflowScheduleInput!) {
+          createWorkflowSchedule(input: $createWorkflowScheduleInput) {
+            schedule {
+              id
+            }
+          }
+        }
+      GRAPHQL
+
+      CREATE_TASK_SCHEDULE = <<-'GRAPHQL'
+        mutation ($createTaskScheduleInput: CreateTaskScheduleInput!) {
+          createTaskSchedule(input: $createTaskScheduleInput) {
+            schedule {
+              id
+            }
+          }
+        }
+      GRAPHQL
+
+      def initialize(http:)
+        @http = http
+      end
+
+      # Makes a GRAPHQL request with some data and sets the correct headers
+      #
+      # @param url [String] the url for the request
+      # @param body [Hash] the payload to send with the request
+      # @return [Hash] the parsed json response
+      def request(url, query, variables = nil, headers = {})
+        body = { 'query' => query }
+        body['variables'] = variables if variables
+
+        res_body = @http.post(url, body, headers)
+        handle_response_body(res_body)
+      end
+
+      private
+
+      def handle_response_body(response_body)
+        if external_error?(response_body)
+          raise Zenaton::ExternalError, format_external_error(response_body)
+        end
+
+        response_body && response_body['data']
+      end
+
+      def external_error?(response_body)
+        response_body&.key?('errors')
+      end
+
+      def format_external_error(response_body)
+        response_body['errors'].map do |error|
+          path = error['path'] ? "- #{error['path']}: " : ''
+          "#{path}#{error['message']}"
+        end.join("\n")
+      end
+    end
+  end
+end

lib/zenaton/services/http.rb

@@ -39,23 +39,25 @@ def get(url)
       #
       # @param url [String] the url for the request
       # @param body [Hash] the payload to send with the request
+      # @param headers [Hash] additional headers to send with the request
       # @return [Hash] the parsed json response
-      def post(url, body)
+      def post(url, body, headers = {})
         parsed_url = parse_url(url)
         request = Net::HTTP::Post.new(parsed_url[:uri])
-        set_body_and_headers(request, post_options(body))
+        set_body_and_headers(request, post_options(body, headers))
         make_request(request, parsed_url)
       end
 
       # Makes a PUT request with some data and sets the correct headers
       #
       # @param url [String] the url for the request
       # @param body [Hash] the payload to send with the request
+      # @param headers [Hash] additional headers to send with the request
       # @return [Hash] the parsed json response
-      def put(url, body)
+      def put(url, body, headers = {})
         parsed_url = parse_url(url)
         request = Net::HTTP::Put.new(parsed_url[:uri])
-        set_body_and_headers(request, put_options(body))
+        set_body_and_headers(request, put_options(body, headers))
         make_request(request, parsed_url)
       end
 
@@ -106,13 +108,14 @@ def default_options
         }
       end
 
-      def post_options(body)
+      def post_options(body, headers)
+        default_post_headers = {
+          'Accept' => 'application/json',
+          'Content-Type' => 'application/json'
+        }
         {
           body: body,
-          headers: {
-            'Accept' => 'application/json',
-            'Content-Type' => 'application/json'
-          }
+          headers: default_post_headers.merge(headers)
         }
       end
       alias put_options post_options

lib/zenaton/traits/zenatonable.rb

@@ -22,6 +22,15 @@ def dispatch
         Engine.instance.dispatch([self])
       end
 
+      def schedule(cron)
+        if !cron.is_a?(String) || cron.blank?
+          raise InvalidArgumentError,
+                "The cron passed to 'schedule' must be a non empty string"
+        end
+
+        Engine.instance.schedule([self], cron)
+      end
+
       class_methods do
         # Search for workflows to interact with.
         # For available methods, see {Zenaton::Query::Builder}