{"action":"created","comment":{"url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/comments/79715089","id":79715089,"diff_hunk":"@@ -330,64 +278,76 @@\n    handler-fn :- message-fn-schema]\n   (swap! listener-atom conj [pred handler-fn]))\n \n-(defn start-receiver\n-  [connection endpoint discard-dir process-msg]\n-  (let [sess (.createSession connection true 0)\n-        q (.createQueue sess endpoint)\n-        consumer (.createConsumer sess q)\n-        producer (.createProducer sess q)\n-        send #(mq/commit-or-rollback sess\n-                (.send producer (mq/to-jms-message sess %1 %2)))\n-        handle (create-message-handler send discard-dir process-msg)]\n-    (.setMessageListener\n-     consumer\n-     (reify MessageListener\n-       (onMessage [this msg]\n-         (try\n-           (mq/commit-or-rollback sess (handle (mq/convert-jms-message msg)))\n-           (catch Throwable ex\n-             (log/error ex \"message receive failed\")\n-             (throw ex))))))\n-    {:session sess :consumer consumer :producer producer}))\n+(def one-hour (* 1000 60 60))\n+\n+(defn send-delayed-message [command-chan delay-pool]\n+  (fn [cmd]\n+    (let [narrowed-entry (dissoc cmd :payload)]\n+      (after one-hour #(async/>!! command-chan narrowed-entry) delay-pool))))\n+\n+(defn create-command-consumer\n+  \"Create and return a command handler. This function does the work of\n+  consuming/storing a command. Handled commands are acknowledged here\"\n+  [q command-chan dlo delay-pool command-handler]\n+  (let [handle-message (message-handler q\n+                                        dlo\n+                                        (send-delayed-message command-chan\n+                                                              delay-pool)\n+                                        command-handler)]\n+    (fn [message]\n+      ;; When the queue is shutting down, it sends nil message\n+      (when message\n+        (try\n+          (handle-message message)\n+          (catch Exception ex\n+            (log/error ex \"Unable to process message. Message not acknowledged and will be retried\")))))))\n+\n+(defn create-command-handler-threadpool\n+  \"Creates an unbounded threadpool with the intent that access to the\n+  threadpool is bounded by the semaphore. Implicitly the threadpool is\n+  bounded by `size`, but since the semaphore is handling that aspect,\n+  it's more efficient to use an unbounded pool and not duplicate the\n+  constraint in both the semaphore and the threadpool\"\n+  [size]\n+  (gtp/create-threadpool size \"cmd-proc-thread-%d\" 10000))","path":"src/puppetlabs/puppetdb/mq_listener.clj","position":403,"original_position":403,"commit_id":"37022a83369578a9642b3972044863a1db286183","original_commit_id":"7c0b89d29ec81c026b76816de76587ab0b6a5250","user":{"login":"senior","id":215910,"avatar_url":"https://avatars.githubusercontent.com/u/215910?v=3","gravatar_id":"","url":"https://api.github.com/users/senior","html_url":"https://github.com/senior","followers_url":"https://api.github.com/users/senior/followers","following_url":"https://api.github.com/users/senior/following{/other_user}","gists_url":"https://api.github.com/users/senior/gists{/gist_id}","starred_url":"https://api.github.com/users/senior/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/senior/subscriptions","organizations_url":"https://api.github.com/users/senior/orgs","repos_url":"https://api.github.com/users/senior/repos","events_url":"https://api.github.com/users/senior/events{/privacy}","received_events_url":"https://api.github.com/users/senior/received_events","type":"User","site_admin":false},"body":"No lost messages. Depending on the timing it could mean abandoning a transaction in the middle. It could mean the transaction was committed, but we haven't removed the message from disk yet (we'd reprocess it on restart) etc.","created_at":"2016-09-20T21:19:31Z","updated_at":"2016-09-20T21:19:31Z","html_url":"https://github.com/puppetlabs/puppetdb/pull/2061#discussion_r79715089","pull_request_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061","_links":{"self":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/comments/79715089"},"html":{"href":"https://github.com/puppetlabs/puppetdb/pull/2061#discussion_r79715089"},"pull_request":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061"}}},"pull_request":{"url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061","id":84892846,"html_url":"https://github.com/puppetlabs/puppetdb/pull/2061","diff_url":"https://github.com/puppetlabs/puppetdb/pull/2061.diff","patch_url":"https://github.com/puppetlabs/puppetdb/pull/2061.patch","issue_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/2061","number":2061,"state":"open","locked":false,"title":"Migrate from ActiveMQ to Stockpile","user":{"login":"senior","id":215910,"avatar_url":"https://avatars.githubusercontent.com/u/215910?v=3","gravatar_id":"","url":"https://api.github.com/users/senior","html_url":"https://github.com/senior","followers_url":"https://api.github.com/users/senior/followers","following_url":"https://api.github.com/users/senior/following{/other_user}","gists_url":"https://api.github.com/users/senior/gists{/gist_id}","starred_url":"https://api.github.com/users/senior/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/senior/subscriptions","organizations_url":"https://api.github.com/users/senior/orgs","repos_url":"https://api.github.com/users/senior/repos","events_url":"https://api.github.com/users/senior/events{/privacy}","received_events_url":"https://api.github.com/users/senior/received_events","type":"User","site_admin":false},"body":"This PR merges the cmd-proc-refactor branch into master and moves from ActiveMQ to Stockpile. Some parts of this PR have been reviewed in prior PRs (before going into cmd-proc-refactor) and others have not. The code that has been reviewed, often had a caveat of, \"this is good enough for now\" or \"this will change in the next set of commits\" etc. So this review is intended to cover those situations.\r\n\r\nIn particular, I'm looking for review comments to be in one of the three categories.\r\n\r\n1. Things that are obviously bad/wrong/inconsistent and that need to be fixed/reworked/cleaned up before we merge into master.\r\n\r\n2. Significant changes/refactoring that could/should be done soon but after we merge to master (before we cut a release). An example of this is merging the mq_listener and command namespaces (ticketed at [PDB-3038](https://tickets.puppetlabs.com/browse/PDB-3038)).\r\n\r\n3. Significant changes/refactoring that are enabled by this change, but not critical to get in before we release. We'll ticket these and put them in the backlog.\r\n\r\nBelow are the changes we know need to be made before releasing, but were planning to work on after merging to master\r\n\r\n- We currently store commands with a filename <integer>_<received_time>_<command>_<version>_<certname>.json where command is our normal command names (i.e. \"replace catalog\"). We should have something more concise and without spaces [PDB-2982](https://tickets.puppetlabs.com/browse/PDB-2982)\r\n\r\n- Deprecate old ActiveMQ and DLO related config that is no longer relevant [PDB-2948](https://tickets.puppetlabs.com/browse/PDB-2948)\r\n\r\n- Ensure the new stockpile code is HUPable [PDB-2983](https://tickets.puppetlabs.com/browse/PDB-2983)\r\n\r\n- Ensure that in the unlikely case that we have a hugely long certname, that we gracefully handle filenames that are too long [PDB-3039](https://tickets.puppetlabs.com/browse/PDB-3039)\r\n\r\n- Merge the mq_listener and command namespaces [PDB-3038](https://tickets.puppetlabs.com/browse/PDB-3038)\r\n\r\nThere are several performance enhancements we're planning to make before a release, but those should be independent of this set of core functionality.","created_at":"2016-09-12T15:32:32Z","updated_at":"2016-09-20T21:19:31Z","closed_at":null,"merged_at":null,"merge_commit_sha":"400ab941e0ea7ce55d01e16c417e7f8529b4a1b4","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061/commits","review_comments_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061/comments","review_comment_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/comments{/number}","comments_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/2061/comments","statuses_url":"https://api.github.com/repos/puppetlabs/puppetdb/statuses/37022a83369578a9642b3972044863a1db286183","head":{"label":"puppetlabs:cmd-proc-refactor","ref":"cmd-proc-refactor","sha":"37022a83369578a9642b3972044863a1db286183","user":{"login":"puppetlabs","id":234268,"avatar_url":"https://avatars.githubusercontent.com/u/234268?v=3","gravatar_id":"","url":"https://api.github.com/users/puppetlabs","html_url":"https://github.com/puppetlabs","followers_url":"https://api.github.com/users/puppetlabs/followers","following_url":"https://api.github.com/users/puppetlabs/following{/other_user}","gists_url":"https://api.github.com/users/puppetlabs/gists{/gist_id}","starred_url":"https://api.github.com/users/puppetlabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/puppetlabs/subscriptions","organizations_url":"https://api.github.com/users/puppetlabs/orgs","repos_url":"https://api.github.com/users/puppetlabs/repos","events_url":"https://api.github.com/users/puppetlabs/events{/privacy}","received_events_url":"https://api.github.com/users/puppetlabs/received_events","type":"Organization","site_admin":false},"repo":{"id":2193032,"name":"puppetdb","full_name":"puppetlabs/puppetdb","owner":{"login":"puppetlabs","id":234268,"avatar_url":"https://avatars.githubusercontent.com/u/234268?v=3","gravatar_id":"","url":"https://api.github.com/users/puppetlabs","html_url":"https://github.com/puppetlabs","followers_url":"https://api.github.com/users/puppetlabs/followers","following_url":"https://api.github.com/users/puppetlabs/following{/other_user}","gists_url":"https://api.github.com/users/puppetlabs/gists{/gist_id}","starred_url":"https://api.github.com/users/puppetlabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/puppetlabs/subscriptions","organizations_url":"https://api.github.com/users/puppetlabs/orgs","repos_url":"https://api.github.com/users/puppetlabs/repos","events_url":"https://api.github.com/users/puppetlabs/events{/privacy}","received_events_url":"https://api.github.com/users/puppetlabs/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/puppetlabs/puppetdb","description":"Centralized Puppet Storage","fork":false,"url":"https://api.github.com/repos/puppetlabs/puppetdb","forks_url":"https://api.github.com/repos/puppetlabs/puppetdb/forks","keys_url":"https://api.github.com/repos/puppetlabs/puppetdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/puppetlabs/puppetdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/puppetlabs/puppetdb/teams","hooks_url":"https://api.github.com/repos/puppetlabs/puppetdb/hooks","issue_events_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/events{/number}","events_url":"https://api.github.com/repos/puppetlabs/puppetdb/events","assignees_url":"https://api.github.com/repos/puppetlabs/puppetdb/assignees{/user}","branches_url":"https://api.github.com/repos/puppetlabs/puppetdb/branches{/branch}","tags_url":"https://api.github.com/repos/puppetlabs/puppetdb/tags","blobs_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/puppetlabs/puppetdb/statuses/{sha}","languages_url":"https://api.github.com/repos/puppetlabs/puppetdb/languages","stargazers_url":"https://api.github.com/repos/puppetlabs/puppetdb/stargazers","contributors_url":"https://api.github.com/repos/puppetlabs/puppetdb/contributors","subscribers_url":"https://api.github.com/repos/puppetlabs/puppetdb/subscribers","subscription_url":"https://api.github.com/repos/puppetlabs/puppetdb/subscription","commits_url":"https://api.github.com/repos/puppetlabs/puppetdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/puppetlabs/puppetdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/puppetlabs/puppetdb/contents/{+path}","compare_url":"https://api.github.com/repos/puppetlabs/puppetdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/puppetlabs/puppetdb/merges","archive_url":"https://api.github.com/repos/puppetlabs/puppetdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/puppetlabs/puppetdb/downloads","issues_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues{/number}","pulls_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls{/number}","milestones_url":"https://api.github.com/repos/puppetlabs/puppetdb/milestones{/number}","notifications_url":"https://api.github.com/repos/puppetlabs/puppetdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/puppetlabs/puppetdb/labels{/name}","releases_url":"https://api.github.com/repos/puppetlabs/puppetdb/releases{/id}","deployments_url":"https://api.github.com/repos/puppetlabs/puppetdb/deployments","created_at":"2011-08-11T18:54:46Z","updated_at":"2016-09-13T10:21:50Z","pushed_at":"2016-09-20T20:41:57Z","git_url":"git://github.com/puppetlabs/puppetdb.git","ssh_url":"git@github.com:puppetlabs/puppetdb.git","clone_url":"https://github.com/puppetlabs/puppetdb.git","svn_url":"https://github.com/puppetlabs/puppetdb","homepage":"http://docs.puppetlabs.com/puppetdb","size":20366,"stargazers_count":194,"watchers_count":194,"language":"Clojure","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":161,"mirror_url":null,"open_issues_count":5,"forks":161,"open_issues":5,"watchers":194,"default_branch":"master"}},"base":{"label":"puppetlabs:master","ref":"master","sha":"9734ef199b66c755adf4eb718cea42aa900f4248","user":{"login":"puppetlabs","id":234268,"avatar_url":"https://avatars.githubusercontent.com/u/234268?v=3","gravatar_id":"","url":"https://api.github.com/users/puppetlabs","html_url":"https://github.com/puppetlabs","followers_url":"https://api.github.com/users/puppetlabs/followers","following_url":"https://api.github.com/users/puppetlabs/following{/other_user}","gists_url":"https://api.github.com/users/puppetlabs/gists{/gist_id}","starred_url":"https://api.github.com/users/puppetlabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/puppetlabs/subscriptions","organizations_url":"https://api.github.com/users/puppetlabs/orgs","repos_url":"https://api.github.com/users/puppetlabs/repos","events_url":"https://api.github.com/users/puppetlabs/events{/privacy}","received_events_url":"https://api.github.com/users/puppetlabs/received_events","type":"Organization","site_admin":false},"repo":{"id":2193032,"name":"puppetdb","full_name":"puppetlabs/puppetdb","owner":{"login":"puppetlabs","id":234268,"avatar_url":"https://avatars.githubusercontent.com/u/234268?v=3","gravatar_id":"","url":"https://api.github.com/users/puppetlabs","html_url":"https://github.com/puppetlabs","followers_url":"https://api.github.com/users/puppetlabs/followers","following_url":"https://api.github.com/users/puppetlabs/following{/other_user}","gists_url":"https://api.github.com/users/puppetlabs/gists{/gist_id}","starred_url":"https://api.github.com/users/puppetlabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/puppetlabs/subscriptions","organizations_url":"https://api.github.com/users/puppetlabs/orgs","repos_url":"https://api.github.com/users/puppetlabs/repos","events_url":"https://api.github.com/users/puppetlabs/events{/privacy}","received_events_url":"https://api.github.com/users/puppetlabs/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/puppetlabs/puppetdb","description":"Centralized Puppet Storage","fork":false,"url":"https://api.github.com/repos/puppetlabs/puppetdb","forks_url":"https://api.github.com/repos/puppetlabs/puppetdb/forks","keys_url":"https://api.github.com/repos/puppetlabs/puppetdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/puppetlabs/puppetdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/puppetlabs/puppetdb/teams","hooks_url":"https://api.github.com/repos/puppetlabs/puppetdb/hooks","issue_events_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/events{/number}","events_url":"https://api.github.com/repos/puppetlabs/puppetdb/events","assignees_url":"https://api.github.com/repos/puppetlabs/puppetdb/assignees{/user}","branches_url":"https://api.github.com/repos/puppetlabs/puppetdb/branches{/branch}","tags_url":"https://api.github.com/repos/puppetlabs/puppetdb/tags","blobs_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/puppetlabs/puppetdb/statuses/{sha}","languages_url":"https://api.github.com/repos/puppetlabs/puppetdb/languages","stargazers_url":"https://api.github.com/repos/puppetlabs/puppetdb/stargazers","contributors_url":"https://api.github.com/repos/puppetlabs/puppetdb/contributors","subscribers_url":"https://api.github.com/repos/puppetlabs/puppetdb/subscribers","subscription_url":"https://api.github.com/repos/puppetlabs/puppetdb/subscription","commits_url":"https://api.github.com/repos/puppetlabs/puppetdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/puppetlabs/puppetdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/puppetlabs/puppetdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/puppetlabs/puppetdb/contents/{+path}","compare_url":"https://api.github.com/repos/puppetlabs/puppetdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/puppetlabs/puppetdb/merges","archive_url":"https://api.github.com/repos/puppetlabs/puppetdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/puppetlabs/puppetdb/downloads","issues_url":"https://api.github.com/repos/puppetlabs/puppetdb/issues{/number}","pulls_url":"https://api.github.com/repos/puppetlabs/puppetdb/pulls{/number}","milestones_url":"https://api.github.com/repos/puppetlabs/puppetdb/milestones{/number}","notifications_url":"https://api.github.com/repos/puppetlabs/puppetdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/puppetlabs/puppetdb/labels{/name}","releases_url":"https://api.github.com/repos/puppetlabs/puppetdb/releases{/id}","deployments_url":"https://api.github.com/repos/puppetlabs/puppetdb/deployments","created_at":"2011-08-11T18:54:46Z","updated_at":"2016-09-13T10:21:50Z","pushed_at":"2016-09-20T20:41:57Z","git_url":"git://github.com/puppetlabs/puppetdb.git","ssh_url":"git@github.com:puppetlabs/puppetdb.git","clone_url":"https://github.com/puppetlabs/puppetdb.git","svn_url":"https://github.com/puppetlabs/puppetdb","homepage":"http://docs.puppetlabs.com/puppetdb","size":20366,"stargazers_count":194,"watchers_count":194,"language":"Clojure","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":161,"mirror_url":null,"open_issues_count":5,"forks":161,"open_issues":5,"watchers":194,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061"},"html":{"href":"https://github.com/puppetlabs/puppetdb/pull/2061"},"issue":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/issues/2061"},"comments":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/issues/2061/comments"},"review_comments":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061/comments"},"review_comment":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/pulls/2061/commits"},"statuses":{"href":"https://api.github.com/repos/puppetlabs/puppetdb/statuses/37022a83369578a9642b3972044863a1db286183"}}}}
{"actor":{"display_login":"senior"}}