das/src/agents/query_engine/PatternMatchingQueryProcessor.h at 4f4dd67ec4993e7e83376199ef8427b213ee25cf · singnet/das · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
#pragma once

#include <map>
#include <memory>
#include <stack>
#include <thread>

#include "AtomDBSingleton.h"
#include "BusCommandProcessor.h"
#include "PatternMatchingQueryProxy.h"
#include "QueryElement.h"
#include "Sink.h"
#include "StoppableThread.h"

using namespace std;
using namespace service_bus;
using namespace query_element;

namespace atomdb {

/**
 * Bus element responsible for processing PATTERN_MATCHING_QUERY commands.
 */
class PatternMatchingQueryProcessor : public BusCommandProcessor {
   public:
    PatternMatchingQueryProcessor();
    ~PatternMatchingQueryProcessor();

    // ---------------------------------------------------------------------------------------------
    // Virtual BusCommandProcessor API

    /**
     * Returns an empty instance of the PatternMatchingQueryProxy.
     *
     * @return An empty instance of the PatternMatchingQueryProxy.
     */
    virtual shared_ptr<BusCommandProxy> factory_empty_proxy();

    /**
     * Method which is called when a command owned by this processor is issued in the bus.
     */
    virtual void run_command(shared_ptr<BusCommandProxy> proxy);

   private:
    void update_attention_broker_single_answer(shared_ptr<PatternMatchingQueryProxy> proxy,
                                               QueryAnswer* answer,
                                               set<string>& joint_answer);
    void update_attention_broker_joint_answer(shared_ptr<PatternMatchingQueryProxy> proxy,
                                              set<string>& joint_answer);
    void process_query_answers(shared_ptr<PatternMatchingQueryProxy> proxy,
                               shared_ptr<Sink> query_sink,
                               set<string>& joint_answer,
                               unsigned int& answer_count);
    shared_ptr<QueryElement> parse_metta_query(shared_ptr<PatternMatchingQueryProxy> proxy);
    shared_ptr<QueryElement> setup_query_tree(shared_ptr<PatternMatchingQueryProxy> proxy);
    void thread_process_one_query(shared_ptr<StoppableThread>,
                                  shared_ptr<PatternMatchingQueryProxy> proxy);
    shared_ptr<QueryElement> build_link_template(shared_ptr<PatternMatchingQueryProxy> proxy,
                                                 unsigned int cursor,
                                                 stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_link_template2(shared_ptr<PatternMatchingQueryProxy> proxy,
                                                  unsigned int cursor,
                                                  stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_and(shared_ptr<PatternMatchingQueryProxy> proxy,
                                       unsigned int cursor,
                                       stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_or(shared_ptr<PatternMatchingQueryProxy> proxy,
                                      unsigned int cursor,
                                      stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_chain(shared_ptr<PatternMatchingQueryProxy> proxy,
                                         unsigned int cursor,
                                         stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_link(shared_ptr<PatternMatchingQueryProxy> proxy,
                                        unsigned int cursor,
                                        stack<shared_ptr<QueryElement>>& element_stack);

    shared_ptr<QueryElement> build_unique_assignment_filter(
        shared_ptr<PatternMatchingQueryProxy> proxy,
        unsigned int cursor,
        stack<shared_ptr<QueryElement>>& element_stack);

    void remove_query_thread(const string& stoppable_thread_id);

    map<string, shared_ptr<StoppableThread>> query_threads;
    mutex query_threads_mutex;
    shared_ptr<PatternMatchingQueryProxy> proxy;
    shared_ptr<AtomDB> atomdb;
    static string AND;
    static string OR;
    static string CHAIN;
};

}  // namespace atomdb