def _start_event_dispatch(self):
        """
        Registers the event listener by using a URL that is composed from
        CFG.server.oms.host, CFG.server.oms.port, and CFG.server.oms.path.

        NOTE: the same listener URL will be registered by multiple RSN platform
        drivers. See other related notes in this file.

        @see https://jira.oceanobservatories.org/tasks/browse/OOIION-1287
        @see https://jira.oceanobservatories.org/tasks/browse/OOIION-968
        """

        # gateway host and port to compose URL:
        host = CFG.get_safe('server.oms.host', "localhost")
        port = CFG.get_safe('server.oms.port', "5000")
        path = CFG.get_safe('server.oms.path', "/ion-service/oms_event")

        self.listener_url = "http://%s:%s%s" % (host, port, path)
        self._register_event_listener(self.listener_url)

        # start OMSDeviceStatusEvent listener to notify the agent about those:
        def event_received(evt, *args, **kwargs):
            log.debug('%r: OmsEventListener received: %s', self._platform_id, evt)
            self._send_event(ExternalEventDriverEvent(evt))

        self._event_listener = self._create_event_subscriber(
            event_type   = 'OMSDeviceStatusEvent',
            origin       = self._platform_id,
            origin_type  = 'OMS Platform',
            callback     = event_received)

        log.debug("%r: started OMSDeviceStatusEvent listener", self._platform_id)

        return "OK"

    def on_init(self):
        log.info("Ingestion starting")
        self.exchange_name = "ingestion_process"

        plugin_cls = CFG.get_safe(CONFIG_KEY + ".plugin")
        self.plugin = named_any(plugin_cls)(self)
        log.info("Started ingestion plugin '%s'", plugin_cls)

        self.persistence_formats = {}
        self.persistence_objects = {}
        self.default_persistence_format = CFG.get_safe(CONFIG_KEY + ".persist.persistence_format")
        self._require_persistence_layer(self.default_persistence_format)

        self.stream_sub = StreamSubscriber(process=self, exchange_name=self.exchange_name,
                                           callback=self.process_package)
        streams = CFG.get_safe(CONFIG_KEY + ".stream_subscriptions") or []
        for stream in streams:
            if isinstance(stream, list):
                stream = StreamRoute(exchange_point=stream[0], routing_key=stream[1])

            log.info("Ingestion subscribed to stream '%s'", stream)
            self.stream_sub.add_stream_subscription(stream)

        self.plugin.on_init()

        self.stream_sub.start()

    def _start_event_dispatch(self):
        """
        Registers the event listener by using a URL that is composed from
        CFG.server.oms.host, CFG.server.oms.port, and CFG.server.oms.path.

        NOTE: the same listener URL will be registered by multiple RSN platform
        drivers. See other related notes in this file.

        @see https://jira.oceanobservatories.org/tasks/browse/OOIION-1287
        @see https://jira.oceanobservatories.org/tasks/browse/OOIION-968
        """

        # gateway host and port to compose URL:
        host = CFG.get_safe('server.oms.host', "localhost")
        port = CFG.get_safe('server.oms.port', "5000")
        path = CFG.get_safe('server.oms.path', "/ion-service/oms_event")

        #the above are defined in pyon.cfg
        #we will override local host for debugging inside the VM
        host = "10.208.79.19"

        self.listener_url = "http://%s:%s%s" % (host, port, path)
        self._register_event_listener(self.listener_url)

        return "OK"

    def do_auto_preload(self):
        """ Load configured preload scenarios and remembers in directory which ones
        have been loaded. """
        log.info("Executing auto preload")
        preload_entries = self.process.container.directory.lookup("/System/Preload")
        preload_changed = False
        if preload_entries is None:
            preload_entries = dict(scenarios={})
            preload_changed = True

        # "HACK" Preload core to get app specifics in
        if "core" not in preload_entries["scenarios"]:
            preload_entries["scenarios"]["core"] = self._get_dir_entry("core")
            self.do_preload_core()
            preload_changed = True

        preload_scenarios = CFG.get_safe("scion.preload.scenarios") or []
        preload_scope = CFG.get_safe("scion.preload.scope")
        for scenario_info in preload_scenarios:
            scope, scenario = scenario_info
            if scope == "all" or scope == preload_scope:
                if scenario not in preload_entries["scenarios"]:
                    preload_entries["scenarios"][scenario] = self._get_dir_entry(scenario)
                    changed = self.do_preload_master(scenario)
                    preload_changed = preload_changed or changed

        if preload_changed:
            self.process.container.directory.register("/System", "Preload", scenarios=preload_entries["scenarios"])

def setting_up_smtp_client():
    '''
    Sets up the smtp client
    '''

    #------------------------------------------------------------------------------------
    # the default smtp server
    #------------------------------------------------------------------------------------

    ION_SMTP_SERVER = 'mail.oceanobservatories.org'

    smtp_host = CFG.get_safe('server.smtp.host', ION_SMTP_SERVER)
    smtp_port = CFG.get_safe('server.smtp.port', 25)
    smtp_sender = CFG.get_safe('server.smtp.sender')
    smtp_password = CFG.get_safe('server.smtp.password')

    if CFG.get_safe('system.smtp',False): #Default is False - use the fake_smtp
        log.debug('Using the real SMTP library to send email notifications!')

        smtp_client = smtplib.SMTP(smtp_host)
        smtp_client.ehlo()
        smtp_client.starttls()
        smtp_client.login(smtp_sender, smtp_password)

    else:
        log.debug('Using a fake SMTP library to simulate email notifications!')

        smtp_client = fake_smtplib.SMTP(smtp_host)

    return smtp_client

    def setUp(self):
        super(DiscoveryIntTest, self).setUp()

        self._start_container()
        self.addCleanup(DiscoveryIntTest.es_cleanup)
        self.container.start_rel_from_url('res/deploy/r2dm.yml')

        self.discovery = DiscoveryServiceClient()
        self.catalog   = CatalogManagementServiceClient()
        self.ims       = IndexManagementServiceClient()
        self.rr        = ResourceRegistryServiceClient()

        if use_es:
            self.es_host   = CFG.get_safe('server.elasticsearch.host', 'localhost')
            self.es_port   = CFG.get_safe('server.elasticsearch.port', '9200')
            CFG.server.elasticsearch.shards         = 1
            CFG.server.elasticsearch.replicas       = 0
            CFG.server.elasticsearch.river_shards   = 1
            CFG.server.elasticsearch.river_replicas = 0
            self.es = ep.ElasticSearch(
                host=self.es_host,
                port=self.es_port,
                timeout=10,
                verbose=True
            )
            op = DotDict(CFG)
            op.op = 'clean_bootstrap'
            self.container.spawn_process('index_bootstrap','ion.processes.bootstrap.index_bootstrap','IndexBootStrap', op)

def setting_up_smtp_client():
    """
    Sets up the smtp client
    """

    #------------------------------------------------------------------------------------
    # the default smtp server
    #------------------------------------------------------------------------------------
#    smtp_client = None
    smtp_host = CFG.get_safe('server.smtp.host')
    smtp_port = CFG.get_safe('server.smtp.port', 25)
#    smtp_sender = CFG.get_safe('server.smtp.sender')
#    smtp_password = CFG.get_safe('server.smtp.password')

    if CFG.get_safe('system.smtp',False): #Default is False - use the fake_smtp
        log.debug('Using the real SMTP library to send email notifications! host = %s', smtp_host)

#        smtp_client = smtplib.SMTP(smtp_host)
#        smtp_client.ehlo()
#        smtp_client.starttls()
#        smtp_client.login(smtp_sender, smtp_password)

        smtp_client = smtplib.SMTP(smtp_host, smtp_port)
        log.debug("In setting up smtp client using the smtp client: %s", smtp_client)
        log.debug("Message received after ehlo exchange: %s", str(smtp_client.ehlo()))
#        smtp_client.login(smtp_sender, smtp_password)
    else:
        log.debug('Using a fake SMTP library to simulate email notifications!')

        smtp_client = fake_smtplib.SMTP(smtp_host)

    return smtp_client

    def test_pydap(self):
        if not CFG.get_safe('bootstrap.use_pydap',False):
            raise unittest.SkipTest('PyDAP is off (bootstrap.use_pydap)')
        ph = ParameterHelper(self.dataset_management, self.addCleanup)
        pdict_id = ph.create_extended_parsed()

        stream_def_id = self.pubsub_management.create_stream_definition('example', parameter_dictionary_id=pdict_id)
        self.addCleanup(self.pubsub_management.delete_stream_definition, stream_def_id)

        tdom, sdom = time_series_domain()

        dp = DataProduct(name='example')
        dp.spatial_domain = sdom.dump()
        dp.temporal_domain = tdom.dump()

        data_product_id = self.data_product_management.create_data_product(dp, stream_def_id)
        self.addCleanup(self.data_product_management.delete_data_product, data_product_id)
        
        self.data_product_management.activate_data_product_persistence(data_product_id)
        self.addCleanup(self.data_product_management.suspend_data_product_persistence, data_product_id)

        dataset_id = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)[0][0]
        monitor = DatasetMonitor(dataset_id)
        self.addCleanup(monitor.stop)

        rdt = ph.get_rdt(stream_def_id)
        ph.fill_rdt(rdt,10)
        ph.publish_rdt_to_data_product(data_product_id, rdt)
        self.assertTrue(monitor.event.wait(10))


        gevent.sleep(1) # Yield to other greenlets, had an issue with connectivity

        pydap_host = CFG.get_safe('server.pydap.host','localhost')
        pydap_port = CFG.get_safe('server.pydap.port',8001)
        url = 'http://%s:%s/%s' %(pydap_host, pydap_port, dataset_id)

        ds = open_url(url)
        np.testing.assert_array_equal(ds['time'][:], np.arange(10))
        untested = []
        for k,v in rdt.iteritems():
            if k==rdt.temporal_parameter:
                continue
            context = rdt.context(k)
            if isinstance(context.param_type, QuantityType):
                np.testing.assert_array_equal(ds[k][k][:][0], rdt[k])
            elif isinstance(context.param_type, ArrayType):
                values = np.empty(rdt[k].shape, dtype='O')
                for i,obj in enumerate(rdt[k]):
                    values[i] = str(obj)
                np.testing.assert_array_equal(ds[k][k][:][0], values)
            elif isinstance(context.param_type, ConstantType):
                np.testing.assert_array_equal(ds[k][k][:][0], rdt[k])
            elif isinstance(context.param_type, CategoryType):
                np.testing.assert_array_equal(ds[k][k][:][0], rdt[k])
            else:
                untested.append('%s (%s)' % (k,context.param_type))
        if untested:
            raise AssertionError('Untested parameters: %s' % untested)

    def on_start(self):  # pragma: no cover
        super(DiscoveryService, self).on_start()

        self.use_es = CFG.get_safe("system.elasticsearch", False)

        self.elasticsearch_host = CFG.get_safe("server.elasticsearch.host", "localhost")
        self.elasticsearch_port = CFG.get_safe("server.elasticsearch.port", "9200")

        self.ep = EventPublisher(event_type="SearchBufferExceededEvent")
        self.heuristic_cutoff = 4

    def on_start(self): # pragma no cover
        super(DiscoveryService,self).on_start()

        self.use_es = CFG.get_safe('system.elasticsearch',False)

        self.elasticsearch_host = CFG.get_safe('server.elasticsearch.host','localhost')
        self.elasticsearch_port = CFG.get_safe('server.elasticsearch.port','9200')

        self.ep = EventPublisher(event_type = 'SearchBufferExceededEvent')
        self.heuristic_cutoff = 4

    def es_cleanup():
        es_host = CFG.get_safe("server.elasticsearch.host", "localhost")
        es_port = CFG.get_safe("server.elasticsearch.port", "9200")
        es = ep.ElasticSearch(host=es_host, port=es_port, timeout=10)
        indexes = STD_INDEXES.keys()
        indexes.append("%s_resources_index" % get_sys_name().lower())
        indexes.append("%s_events_index" % get_sys_name().lower())

        for index in indexes:
            IndexManagementService._es_call(es.river_couchdb_delete, index)
            IndexManagementService._es_call(es.index_delete, index)

    def test_ccov_visualization(self):
        '''
        Tests Complex Coverage aggregation of array types and proper splitting of coverages
        tests pydap and the visualization
        '''
        if not CFG.get_safe('bootstrap.use_pydap',False):
            raise unittest.SkipTest('PyDAP is off (bootstrap.use_pydap)')

        data_product_id, stream_def_id = self.make_array_data_product()

        # Make a granule with an array type, give it a few values
        # Send it to google_dt transform, verify output

        rdt = RecordDictionaryTool(stream_definition_id=stream_def_id)
        rdt['time'] = np.arange(2208988800, 2208988810)
        rdt['temp_sample'] = np.arange(10*4).reshape(10,4)
        rdt['cond_sample'] = np.arange(10*4).reshape(10,4)

        dataset_id = self.RR2.find_dataset_id_of_data_product_using_has_dataset(data_product_id)
        dataset_monitor = DatasetMonitor(dataset_id)
        self.addCleanup(dataset_monitor.stop)
        self.ph.publish_rdt_to_data_product(data_product_id, rdt, connection_id='abc1', connection_index='1')
        self.assertTrue(dataset_monitor.event.wait(10))
        dataset_monitor.event.clear()


        rdt = RecordDictionaryTool(stream_definition_id=stream_def_id)
        rdt['time'] = np.arange(2208988810, 2208988820)
        rdt['temp_sample'] = np.arange(10*4).reshape(10,4)
        rdt['cond_sample'] = np.arange(10*4).reshape(10,4)
        self.ph.publish_rdt_to_data_product(data_product_id, rdt, connection_id='abc2', connection_index='1')
        self.assertTrue(dataset_monitor.event.wait(10))
        dataset_monitor.event.clear()

        qstring = '{"stride_time": 1, "parameters": [], "query_type": "highcharts_data", "start_time": 0, "use_direct_access": 0, "end_time": 19}'
        graph = self.visualization.get_visualization_data(data_product_id, qstring)
        self.assertIn('temp_sample[3]', graph)

        granule = self.data_retriever.retrieve(dataset_id)
        rdt = RecordDictionaryTool.load_from_granule(granule)

        np.testing.assert_array_equal(rdt['temp_sample'][0], np.arange(4))

        pydap_host = CFG.get_safe('server.pydap.host','localhost')
        pydap_port = CFG.get_safe('server.pydap.port',8001)
        url = 'http://%s:%s/%s' %(pydap_host, pydap_port, dataset_id)

        ds = open_url(url)

        temp_sample, time = ds['temp_sample']
        temp_values, dim = temp_sample[0]
        np.testing.assert_array_equal(temp_values, np.array(['0.0,1.0,2.0,3.0']))

 def setUp(self):
     self._start_container()
     self.container.start_rel_from_url('res/deploy/r2deploy.yml')
     path = CFG.get_safe('server.pydap.data_path', "RESOURCE:ext/pydap")
     ext_path = FileSystem.get_extended_url(path)
     self.cov,self.filename = _make_coverage(ext_path)
     self.nt = 5
     self.cov.insert_timesteps(self.nt) 
     self.time_data = [i+1 for i in range(self.nt)]
     self.cov.set_parameter_values("time", value=self.time_data)
     host = CFG.get_safe('container.pydap_gateway.web_server.host', 'localhost')
     port = CFG.get_safe('container.pydap_gateway.web_server.port', '8001')
     self.request_url = "http://"+host+":"+str(port)+os.sep+os.path.basename(self.filename)

    def on_init(self):

        # clients
        self.resource_registry = self.container.resource_registry
        self.event_publisher = EventPublisher(OT.NotificationSentEvent)

        # SMTP client configurations
        self.smtp_from = CFG.get_safe('server.smtp.from', '[email protected]')
        self.smtp_host = CFG.get_safe('server.smtp.host', 'localhost')
        self.smtp_port = CFG.get_safe('server.smtp.port', 25)

        # Jinja2 template environment
        self.jinja_env = Environment(loader=FileSystemLoader('res/templates'), trim_blocks=True, lstrip_blocks=True)

        super(NotificationWorker, self).on_init()

    def on_start(self): # pragma no cover
        super(DiscoveryService,self).on_start()

        self.use_es = CFG.get_safe('system.elasticsearch', False)

        self.elasticsearch_host = CFG.get_safe('server.elasticsearch.host','localhost')
        self.elasticsearch_port = CFG.get_safe('server.elasticsearch.port','9200')

        self.ep = EventPublisher(event_type = 'SearchBufferExceededEvent')
        self.heuristic_cutoff = 4

        self.cfg_datastore = CFG.get_safe('container.datastore.default_server', "couchdb")
        self.ds_discovery = None
        if self.cfg_datastore != "couchdb":
            self.ds_discovery = DatastoreDiscovery(self)

def process_query():
    ''' 
    Processes a query from the user
    '''
    elasticsearch_host = CFG.get_safe('server.elasticsearch.host','localhost')
    elasticsearch_port = CFG.get_safe('server.elasticsearch.port','9200')
    try:
        import requests
        data = request.stream.read()
        # pass thru
        r = requests.post('http://%s:%s/_search' % (elasticsearch_host, elasticsearch_port), data=data, headers={'content-type':'application/json'})
        return r.content
    
    except Exception as e:
        return build_error_page(traceback.format_exc())

    def _load_system_actors(self):
        """
        Retrieve system and webauth actors and headers for later use.
        """

        # Retrieve and store system actor and headers.
        system_actor, _ = self.container.resource_registry.find_resources(
            RT.ActorIdentity,
            name=CFG.system.system_actor,
            id_only=False)
        self.system_actor = system_actor[0] if system_actor else None
        self.system_actor_id = system_actor[0]._id if system_actor \
            else 'anonymous'
        self.system_actor_headers = {
            'ion-actor-id': self.system_actor_id,
            'ion-actor-roles': {'ION': ['ION_MANAGER', 'ORG_MANAGER']},
            'expiry':'0'
        }

        # Retrieve and store webauth actor and headers.
        webauth_actor, _ = self.container.resource_registry.find_resources(
            RT.ActorIdentity,
            name=CFG.get_safe("system.web_authentication_actor",
                              "web_authentication"), id_only=False)
        self.webauth_actor = webauth_actor[0] if webauth_actor else None
        self.webauth_actor_id = webauth_actor[0]._id if webauth_actor \
            else 'anonymous'
        self.webauth_actor_headers = {
            'ion-actor-id': self.webauth_actor_id,
            'ion-actor-roles': {'ION': ['ION_MANAGER', 'ORG_MANAGER']},
            'expiry':'0'
        }

 def _get_cov(self, name, nt):
     path = CFG.get_safe('server.pydap.data_path', "RESOURCE:ext/pydap")
     ext_path = FileSystem.get_extended_url(path)
     cov,filename = _make_coverage(ext_path, "the_cov")
     cov.insert_timesteps(nt) 
     cov.set_parameter_values("time", value=nt)
     return cov, filename

    def on_init(self):

        self.http_server = None
        self.server_hostname = self.CFG.get_safe(CFG_PREFIX + '.web_server.hostname', DEFAULT_WEB_SERVER_HOSTNAME)
        self.server_port = self.CFG.get_safe(CFG_PREFIX + '.web_server.port', DEFAULT_WEB_SERVER_PORT)
        self.url_prefix = self.CFG.get_safe(CFG_PREFIX + '.url_prefix') or ""

        self.web_server_enabled = True
        self.logging = None
        self.interaction_observer = None
        self.plugin = None
        app.secret_key = self.__class__.__name__   # Enables sessions (for mscweb)

        #retain a pointer to this object for use in ProcessRPC calls
        global adminui_instance
        adminui_instance = self

        #Start the gevent web server unless disabled
        if self.web_server_enabled:
            self.start_service(self.server_hostname, self.server_port)

            plugin_cls = CFG.get_safe(CFG_PREFIX + '.plugin')
            if plugin_cls:
                cls = named_any(plugin_cls)
                self.plugin = cls(app, self)

    def _gen_service_spec(self, service_name):
        output = {
            "swagger": "2.0",
            "info": {},
            "host": CFG.get_safe("system.web_ui_url", "").rstrip("/").split("//", 1)[1],
            "basePath": CFG.get_safe(UISG_CFG_PREFIX + ".url_prefix"),
            "schemes": ["http"],
        }
        output["info"].update(self.config.get("info", {}))
        if self.config.get("externalDocs", {}):
            output["externalDocs"] = self.config["externalDocs"]

        output["tags"] = self._gen_tags(service_name)
        output["paths"] = self._gen_paths(service_name)

        return output