<reponame>PetreStegaroiu/python-esppy<filename>esppy/connectors/fs.py
#!/usr/bin/env python
# encoding: utf-8
#
# Copyright SAS Institute
#
#  Licensed under the Apache License, Version 2.0 (the License);
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

''' ESP File and Socket Connectors '''

from __future__ import print_function, division, absolute_import, unicode_literals

import numbers
import re
import six
from .base import Connector, prop, map_properties
from ..utils import xml
from ..utils.data import gen_name


class FilePublisher(Connector):
    '''
    Publish events from a file

    Parameters
    ----------
    fsname : string
        The filename or path of the file
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    addcsvflags : string, optional
        Specifies the event type to insert into input CSV events.
        Valid values: 'normal' or 'partialupdate'
    addcsvopcode : string, optional
        Prepends an opcode and comma to input CSV events.
    blocksize : int, optional
        Specifies the number of events to include in a published event block
    cefsyslogprefix : string, optional
        When fstype=cef, specifies that CEF events contain the syslog prefix
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    csvfielddelimiter : string, optional
        Specifies the character delimiter for field data in input CSV events
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    growinginputfile : boolean, optional
        Enables reading from a growing input file by publishers
    header : int, optional
        Specifies the number of input lines to skip before starting
        publish operations.
    ignorecsvparseerrors : boolean, optional
        Specifies that when a field in an input CSV event cannot be parsed,
        the event is dropped, an error is logged, and publishing continues.
    maxevents : int, optional
        Specifies the maximum number of events to publish
    noautogenfield : boolean, optional
        Specifies that input events are missing the key field that is
        autogenerated by the source window.
    prebuffer : boolean, optional
        Controls whether event blocks are buffered to an event block vector
        before doing any injects.
    publishwithupsert : boolean, optional
        Build events with opcode=upsert instead of insert
    rate : int, optional
        Specifies the requested transmit rate in events per second
    repeatcount : int, optional
        Specifies the number of times to repeat the publish operation
    transactional : string, optional
        Sets the event block type to transactional.

    Returns
    -------
    :class:`FilePublisher`

    '''
    connector_key = dict(cls='fs', type='publish')

    property_defs = dict(
        fsname=prop('fsname', dtype='string', required=True),
        fstype=prop('fstype', dtype='string', required=True,
                    default='csv',
                    valid_values=['binary', 'csv', 'xml', 'json',
                                  'syslog', 'hdat', 'cef']),
        snapshot=prop('snapshot', dtype='boolean', required=True, default=False),
        addcsvflags=prop('addcsvflags', dtype='string',
                         valid_values=['normal', '']),
        addcsvopcode=prop('addcsvopcode', dtype='string',
                          valid_values=['insert', '']),
        blocksize=prop('blocksize', dtype='int', valid_expr='value > 0'),
        cefsyslogprefix=prop('blocksize', dtype='string'),
        configfilesection=prop('configfilesection', dtype='string'),
        csvfielddelimiter=prop('csvfielddelimiter', dtype='string'),
        dateformat=prop('dateformat', dtype='string'),
        growinginputfile=prop('growinginputfile', dtype='boolean'),
        header=prop('header', dtype=('boolean', 'string'),
                    valid_values=[True, False, 'full']),
        ignorecsvparseerrors=prop('ignorecsvparseerrors', dtype='boolean'),
        maxevents=prop('maxevents', dtype='int', valid_expr='value >= 0'),
        noautogenfield=prop('noautogenfield', dtype='boolean'),
        prebuffer=prop('prebuffer', dtype='boolean'),
        publish_with_upsert=prop('publishwithupsert', dtype='boolean'),
        rate=prop('rate', dtype='int'),
        repeatcount=prop('repeatcount', dtype='int', valid_expr='value >= 0'),
        transactional=prop('transactional', dtype='string'),
    )

    def __init__(self, fsname=None, fstype=None, name=None, is_active=None,
                 snapshot=None, addcsvflags=None,
                 addcsvopcode=None, blocksize=None, cefsyslogprefix=None,
                 configfilesection=None, csvfielddelimiter=None,
                 dateformat=None,
                 growinginputfile=None, header=None, ignorecsvparseerrors=None,
                 maxevents=None, noautogenfield=None, prebuffer=None,
                 publishwithupsert=None, rate=None, repeatcount=None,
                 transactional=None):
        params = dict(**locals())
        params.pop('self')
        params.pop('is_active')
        name = params.pop('name')
        Connector.__init__(self, 'fs', name=name, type='publish', is_active=is_active,
                           properties=params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class FileSubscriber(Connector):
    '''
    Subscribe to events from a file

    Parameters
    ----------
    fsname : string
        The filename or path of the file
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    collapse : boolean, optional
        Converts UPDATE_BLOCK events to UPDATE events in order to make
        subscriber output publishable.
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    hdatcashostport : string, optional
        Specifies the CAS server host and port
    hdatcaspassword : string, optional
        Specifies the CAS server password
    hdatcasusername : string, optional
        Specifies the CAS server user name
    hdatfilename : string, optional
        Specifies the name of the Objective Analysis Package Data (HDAT)
        file to be written to the Hadoop Distributed File System (HDFS).
    hdatlasrhostport : string, optional
        Specifies the SAS LASR Analytic Server host and port
    hdatlasrkey : string, optional
        Specifies the path to tklasrkey.sh
    hdatmaxdatanodes : int, optional
        Specifies the maximum number of data node connections
    hdatmaxstringlength : int, optional
        Specifies in bytes the fixed size of string fields in Objective
        Analysis Package Data (HDAT) files
    hdatnumthreads : int, optional
        Specifies the size of the thread pool used for multi-threaded
        writes to data node socket connections.
    hdfsblocksize : int, optional
        Specifies in Mbytes the block size used to write an Objective
        Analysis Package Data (HDAT) file.
    hdfsnumreplicas : int, optional
        Specifies the number of Hadoop Distributed File System (HDFS)
        replicas created with writing an Objective Analysis Package
        Data (HDAT) file.
    header : boolean or string, optional
        For a CSV subscriber, specifies to write a header row that
        shows comma-separated fields.
        Valid values: True, False, or 'full' (include opcode flags in header)
    maxfilesize : int, optional
        Specifies the maximum size in bytes of the subscriber output file
    periodicity : int, optional
        Specifies the interval in seconds at which the subscriber output
        file is closed and a new output file opened.
    rate : boolean, optional
        When latency mode is enabled, shows this specified rate in generated
        output files.
    rmretdel : boolean, optional
        Specifies to remove all delete events from event blocks received
        by a subscriber that were introduced by a window retention policy.
    unbufferedoutputstreams : boolean, optional
        Specifies to create an unbuffered stream when writing to a
        file or socket.

    Returns
    -------
    :class:`FileSubscriber`

    '''
    connector_key = dict(cls='fs', type='subscribe')

    property_defs = dict(
        fsname=prop('fsname', dtype='string', required=True),
        fstype=prop('fstype', dtype='string', required=True,
                    default='csv',
                    valid_values=['binary', 'csv', 'xml', 'json',
                                  'syslog', 'hdat', 'cef']),
        snapshot=prop('snapshot', dtype='bool', required=True, default=False),
        collapse=prop('collapse', dtype='bool'),
        configfilesection=prop('configfilesection', dtype='string'),
        dateformat=prop('dateformat', dtype='string'),
        hdatcashostport=prop('hdatcashostport', dtype='string',
                             valid_values=re.compile(r'\w[\w\-\.]*:\d+')),
        hdatcaspassword=prop('hdatcaspassword', dtype='string'),
        hdatcasusername=prop('hdatcasusername', dtype='string'),
        hdatfilename=prop('hdatfilename', dtype='string'),
        hdatlasrhostport=prop('hdatlasrhostport', dtype='string'),
        hdatlasrkey=prop('hdatlasrkey', dtype='string'),
        hdatmaxdatanodes=prop('hdatmaxdatanodes',
                              dtype='int', valid_expr='value > 0'),
        hdatmaxstringlength=prop('hdatmaxstringlength', dtype='int',
                                 valid_expr='value > 0'),
        hdatnumthreads=prop('hdatnumthreads', dtype='int',
                            valid_expr='value >= 0'),
        hdfsblocksize=prop('hdfsblocksize', dtype='int',
                           valid_expr='value >= 0'),
        hdfsnumreplicas=prop('hdfsnumreplicas', dtype='int',
                             valid_expr='value >= 0'),
        header=prop('header', dtype=('bool', 'string'),
                    valid_values=[True, False, 'full']),
        maxfilesize=prop('maxfilesize', dtype='int',
                         valid_expr='value >= 0'),
        periodicity=prop('periodicity', dtype='int',
                         valid_expr='value >= 0'),
        rate=prop('rate', dtype='bool'),
        rmretdel=prop('rmretdel', dtype='bool'),
        unbufferedoutputstreams=prop('unbufferedoutputstreams', dtype='bool'),
    )

    def __init__(self, fsname=None, fstype=None, name=None, is_active=None,
                 snapshot=None, collapse=None, configfilesection=None,
                 dateformat=None, hdatcashostport=None,
                 hdatcaspassword=None, hdatcasusername=None,
                 hdatfilename=None, hdatlasrhostport=None, hdatlasrkey=None,
                 hdatmaxdatanodes=None, hdatmaxstringlength=None,
                 hdatnumthreads=None, hdfsblocksize=None,
                 hdfsnumreplicas=None, header=None, maxfilesize=None,
                 periodicity=None, rate=None, rmretdel=None,
                 unbufferedoutputstreams=None):
        params = dict(**locals())
        params.pop('self')
        params.pop('is_active')
        name = params.pop('name')
        Connector.__init__(self, 'fs', name=name, type='subscribe', is_active=is_active,
                           properties=params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class SocketPublisher(FilePublisher):
    '''
    Publish events from a socket

    Parameters
    ----------
    port : int
        The port number to publish on
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    addcsvflags : string, optional
        Specifies the event type to insert into input CSV events.
        Valid values: 'normal' or 'partialupdate'
    addcsvopcode : string, optional
        Prepends an opcode and comma to input CSV events.
    blocksize : int, optional
        Specifies the number of events to include in a published event block
    cefsyslogprefix : string, optional
        When fstype=cef, specifies that CEF events contain the syslog prefix
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    csvfielddelimiter : string, optional
        Specifies the character delimiter for field data in input CSV events
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    growinginputfile : boolean, optional
        Enables reading from a growing input file by publishers
    header : int, optional
        Specifies the number of input lines to skip before starting
        publish operations.
    ignorecsvparseerrors : boolean, optional
        Specifies that when a field in an input CSV event cannot be parsed,
        the event is dropped, an error is logged, and publishing continues.
    maxevents : int, optional
        Specifies the maximum number of events to publish
    noautogenfield : boolean, optional
        Specifies that input events are missing the key field that is
        autogenerated by the source window.
    prebuffer : boolean, optional
        Controls whether event blocks are buffered to an event block vector
        before doing any injects.
    publishwithupsert : boolean, optional
        Build events with opcode=upsert instead of insert
    rate : int, optional
        Specifies the requested transmit rate in events per second
    repeatcount : int, optional
        Specifies the number of times to repeat the publish operation
    transactional : string, optional
        Sets the event block type to transactional.

    Returns
    -------
    :class:`SocketPublisher`

    '''
    connector_key = dict(cls='fs', type='publish', fsname=re.compile(r':(@\w+@|\d+)$'))

    def __init__(self, port=None, fstype=None, name=None, is_active=None,
                 snapshot=None, addcsvflags=None,
                 addcsvopcode=None, blocksize=None, cefsyslogprefix=None,
                 configfilesection=None, csvfielddelimiter=None,
                 dateformat=None,
                 growinginputfile=None, header=None, ignorecsvparseerrors=None,
                 maxevents=None, noautogenfield=None, prebuffer=None,
                 publishwithupsert=None, rate=None, repeatcount=None,
                 transactional=None):
        params = dict(locals())
        params.pop('is_active')
        if isinstance(port, six.string_types):
            port = port.replace(':', '')
        if port is not None:
            params['fsname'] = ':%s' % int(port)
        params.pop('port', None)
        FilePublisher.__init__(**params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class SocketSubscriber(FileSubscriber):
    '''
    Subscribe to events from a socket

    Parameters
    ----------
    host : string
        The host name where the socket exists
    port : int
        The port number on the server
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    collapse : boolean, optional
        Converts UPDATE_BLOCK events to UPDATE events in order to make
        subscriber output publishable.
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    hdatcashostport : string, optional
        Specifies the CAS server host and port
    hdatcaspassword : string, optional
        Specifies the CAS server password
    hdatcasusername : string, optional
        Specifies the CAS server user name
    hdatfilename : string, optional
        Specifies the name of the Objective Analysis Package Data (HDAT)
        file to be written to the Hadoop Distributed File System (HDFS).
    hdatlasrhostport : string, optional
        Specifies the SAS LASR Analytic Server host and port
    hdatlasrkey : string, optional
        Specifies the path to tklasrkey.sh
    hdatmaxdatanodes : int, optional
        Specifies the maximum number of data node connections
    hdatmaxstringlength : int, optional
        Specifies in bytes the fixed size of string fields in Objective
        Analysis Package Data (HDAT) files
    hdatnumthreads : int, optional
        Specifies the size of the thread pool used for multi-threaded
        writes to data node socket connections.
    hdfsblocksize : int, optional
        Specifies in Mbytes the block size used to write an Objective
        Analysis Package Data (HDAT) file.
    hdfsnumreplicas : int, optional
        Specifies the number of Hadoop Distributed File System (HDFS)
        replicas created with writing an Objective Analysis Package
        Data (HDAT) file.
    header : boolean or string, optional
        For a CSV subscriber, specifies to write a header row that
        shows comma-separated fields.
        Valid values: True, False, or 'full' (include opcode flags in header)
    maxfilesize : int, optional
        Specifies the maximum size in bytes of the subscriber output file
    periodicity : int, optional
        Specifies the interval in seconds at which the subscriber output
        file is closed and a new output file opened.
    rate : boolean, optional
        When latency mode is enabled, shows this specified rate in generated
        output files.
    rmretdel : boolean, optional
        Specifies to remove all delete events from event blocks received
        by a subscriber that were introduced by a window retention policy.
    unbufferedoutputstreams : boolean, optional
        Specifies to create an unbuffered stream when writing to a
        file or socket.

    Returns
    -------
    :class:`SocketSubscriber`

    '''
    connector_key = dict(cls='fs', type='subscribe',
                         fsname=re.compile(r'^(@\w+@|[\w+\.\-]+):(@\w+@|\d+)$'))

    def __init__(self, host=None, port=None, fstype=None, name=None, is_active=None,
                 snapshot=None, collapse=None, configfilesection=None,
                 dateformat=None, hdatcashostport=None,
                 hdatcaspassword=None, hdatcasusername=None,
                 hdatfilename=None, hdatlasrhostport=None, hdatlasrkey=None,
                 hdatmaxdatanodes=None, hdatmaxstringlength=None,
                 hdatnumthreads=None, hdfsblocksize=None, hdfsnumreplicas=None,
                 header=None, maxfilesize=None, periodicity=None,
                 rate=None, rmretdel=None, unbufferedoutputstreams=None):
        params = dict(locals())
        params.pop('is_active')
        if host is not None and port is not None:
            params['fsname'] = '%s:%s' % (host, int(port))
        params.pop('host', None)
        params.pop('port', None)
        FileSubscriber.__init__(**params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        if req[0] is not None:
            host, port = req[0].split(':', 1)
        return cls(host, port, name=name, is_active=is_active, **properties)

#!/usr/bin/env python
# encoding: utf-8
#
# Copyright SAS Institute
#
#  Licensed under the Apache License, Version 2.0 (the License);
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

''' ESP File and Socket Connectors '''

from __future__ import print_function, division, absolute_import, unicode_literals

import numbers
import re
import six
from .base import Connector, prop, map_properties
from ..utils import xml
from ..utils.data import gen_name


class FilePublisher(Connector):
    '''
    Publish events from a file

    Parameters
    ----------
    fsname : string
        The filename or path of the file
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    addcsvflags : string, optional
        Specifies the event type to insert into input CSV events.
        Valid values: 'normal' or 'partialupdate'
    addcsvopcode : string, optional
        Prepends an opcode and comma to input CSV events.
    blocksize : int, optional
        Specifies the number of events to include in a published event block
    cefsyslogprefix : string, optional
        When fstype=cef, specifies that CEF events contain the syslog prefix
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    csvfielddelimiter : string, optional
        Specifies the character delimiter for field data in input CSV events
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    growinginputfile : boolean, optional
        Enables reading from a growing input file by publishers
    header : int, optional
        Specifies the number of input lines to skip before starting
        publish operations.
    ignorecsvparseerrors : boolean, optional
        Specifies that when a field in an input CSV event cannot be parsed,
        the event is dropped, an error is logged, and publishing continues.
    maxevents : int, optional
        Specifies the maximum number of events to publish
    noautogenfield : boolean, optional
        Specifies that input events are missing the key field that is
        autogenerated by the source window.
    prebuffer : boolean, optional
        Controls whether event blocks are buffered to an event block vector
        before doing any injects.
    publishwithupsert : boolean, optional
        Build events with opcode=upsert instead of insert
    rate : int, optional
        Specifies the requested transmit rate in events per second
    repeatcount : int, optional
        Specifies the number of times to repeat the publish operation
    transactional : string, optional
        Sets the event block type to transactional.

    Returns
    -------
    :class:`FilePublisher`

    '''
    connector_key = dict(cls='fs', type='publish')

    property_defs = dict(
        fsname=prop('fsname', dtype='string', required=True),
        fstype=prop('fstype', dtype='string', required=True,
                    default='csv',
                    valid_values=['binary', 'csv', 'xml', 'json',
                                  'syslog', 'hdat', 'cef']),
        snapshot=prop('snapshot', dtype='boolean', required=True, default=False),
        addcsvflags=prop('addcsvflags', dtype='string',
                         valid_values=['normal', '']),
        addcsvopcode=prop('addcsvopcode', dtype='string',
                          valid_values=['insert', '']),
        blocksize=prop('blocksize', dtype='int', valid_expr='value > 0'),
        cefsyslogprefix=prop('blocksize', dtype='string'),
        configfilesection=prop('configfilesection', dtype='string'),
        csvfielddelimiter=prop('csvfielddelimiter', dtype='string'),
        dateformat=prop('dateformat', dtype='string'),
        growinginputfile=prop('growinginputfile', dtype='boolean'),
        header=prop('header', dtype=('boolean', 'string'),
                    valid_values=[True, False, 'full']),
        ignorecsvparseerrors=prop('ignorecsvparseerrors', dtype='boolean'),
        maxevents=prop('maxevents', dtype='int', valid_expr='value >= 0'),
        noautogenfield=prop('noautogenfield', dtype='boolean'),
        prebuffer=prop('prebuffer', dtype='boolean'),
        publish_with_upsert=prop('publishwithupsert', dtype='boolean'),
        rate=prop('rate', dtype='int'),
        repeatcount=prop('repeatcount', dtype='int', valid_expr='value >= 0'),
        transactional=prop('transactional', dtype='string'),
    )

    def __init__(self, fsname=None, fstype=None, name=None, is_active=None,
                 snapshot=None, addcsvflags=None,
                 addcsvopcode=None, blocksize=None, cefsyslogprefix=None,
                 configfilesection=None, csvfielddelimiter=None,
                 dateformat=None,
                 growinginputfile=None, header=None, ignorecsvparseerrors=None,
                 maxevents=None, noautogenfield=None, prebuffer=None,
                 publishwithupsert=None, rate=None, repeatcount=None,
                 transactional=None):
        params = dict(**locals())
        params.pop('self')
        params.pop('is_active')
        name = params.pop('name')
        Connector.__init__(self, 'fs', name=name, type='publish', is_active=is_active,
                           properties=params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class FileSubscriber(Connector):
    '''
    Subscribe to events from a file

    Parameters
    ----------
    fsname : string
        The filename or path of the file
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    collapse : boolean, optional
        Converts UPDATE_BLOCK events to UPDATE events in order to make
        subscriber output publishable.
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    hdatcashostport : string, optional
        Specifies the CAS server host and port
    hdatcaspassword : string, optional
        Specifies the CAS server password
    hdatcasusername : string, optional
        Specifies the CAS server user name
    hdatfilename : string, optional
        Specifies the name of the Objective Analysis Package Data (HDAT)
        file to be written to the Hadoop Distributed File System (HDFS).
    hdatlasrhostport : string, optional
        Specifies the SAS LASR Analytic Server host and port
    hdatlasrkey : string, optional
        Specifies the path to tklasrkey.sh
    hdatmaxdatanodes : int, optional
        Specifies the maximum number of data node connections
    hdatmaxstringlength : int, optional
        Specifies in bytes the fixed size of string fields in Objective
        Analysis Package Data (HDAT) files
    hdatnumthreads : int, optional
        Specifies the size of the thread pool used for multi-threaded
        writes to data node socket connections.
    hdfsblocksize : int, optional
        Specifies in Mbytes the block size used to write an Objective
        Analysis Package Data (HDAT) file.
    hdfsnumreplicas : int, optional
        Specifies the number of Hadoop Distributed File System (HDFS)
        replicas created with writing an Objective Analysis Package
        Data (HDAT) file.
    header : boolean or string, optional
        For a CSV subscriber, specifies to write a header row that
        shows comma-separated fields.
        Valid values: True, False, or 'full' (include opcode flags in header)
    maxfilesize : int, optional
        Specifies the maximum size in bytes of the subscriber output file
    periodicity : int, optional
        Specifies the interval in seconds at which the subscriber output
        file is closed and a new output file opened.
    rate : boolean, optional
        When latency mode is enabled, shows this specified rate in generated
        output files.
    rmretdel : boolean, optional
        Specifies to remove all delete events from event blocks received
        by a subscriber that were introduced by a window retention policy.
    unbufferedoutputstreams : boolean, optional
        Specifies to create an unbuffered stream when writing to a
        file or socket.

    Returns
    -------
    :class:`FileSubscriber`

    '''
    connector_key = dict(cls='fs', type='subscribe')

    property_defs = dict(
        fsname=prop('fsname', dtype='string', required=True),
        fstype=prop('fstype', dtype='string', required=True,
                    default='csv',
                    valid_values=['binary', 'csv', 'xml', 'json',
                                  'syslog', 'hdat', 'cef']),
        snapshot=prop('snapshot', dtype='bool', required=True, default=False),
        collapse=prop('collapse', dtype='bool'),
        configfilesection=prop('configfilesection', dtype='string'),
        dateformat=prop('dateformat', dtype='string'),
        hdatcashostport=prop('hdatcashostport', dtype='string',
                             valid_values=re.compile(r'\w[\w\-\.]*:\d+')),
        hdatcaspassword=prop('hdatcaspassword', dtype='string'),
        hdatcasusername=prop('hdatcasusername', dtype='string'),
        hdatfilename=prop('hdatfilename', dtype='string'),
        hdatlasrhostport=prop('hdatlasrhostport', dtype='string'),
        hdatlasrkey=prop('hdatlasrkey', dtype='string'),
        hdatmaxdatanodes=prop('hdatmaxdatanodes',
                              dtype='int', valid_expr='value > 0'),
        hdatmaxstringlength=prop('hdatmaxstringlength', dtype='int',
                                 valid_expr='value > 0'),
        hdatnumthreads=prop('hdatnumthreads', dtype='int',
                            valid_expr='value >= 0'),
        hdfsblocksize=prop('hdfsblocksize', dtype='int',
                           valid_expr='value >= 0'),
        hdfsnumreplicas=prop('hdfsnumreplicas', dtype='int',
                             valid_expr='value >= 0'),
        header=prop('header', dtype=('bool', 'string'),
                    valid_values=[True, False, 'full']),
        maxfilesize=prop('maxfilesize', dtype='int',
                         valid_expr='value >= 0'),
        periodicity=prop('periodicity', dtype='int',
                         valid_expr='value >= 0'),
        rate=prop('rate', dtype='bool'),
        rmretdel=prop('rmretdel', dtype='bool'),
        unbufferedoutputstreams=prop('unbufferedoutputstreams', dtype='bool'),
    )

    def __init__(self, fsname=None, fstype=None, name=None, is_active=None,
                 snapshot=None, collapse=None, configfilesection=None,
                 dateformat=None, hdatcashostport=None,
                 hdatcaspassword=None, hdatcasusername=None,
                 hdatfilename=None, hdatlasrhostport=None, hdatlasrkey=None,
                 hdatmaxdatanodes=None, hdatmaxstringlength=None,
                 hdatnumthreads=None, hdfsblocksize=None,
                 hdfsnumreplicas=None, header=None, maxfilesize=None,
                 periodicity=None, rate=None, rmretdel=None,
                 unbufferedoutputstreams=None):
        params = dict(**locals())
        params.pop('self')
        params.pop('is_active')
        name = params.pop('name')
        Connector.__init__(self, 'fs', name=name, type='subscribe', is_active=is_active,
                           properties=params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class SocketPublisher(FilePublisher):
    '''
    Publish events from a socket

    Parameters
    ----------
    port : int
        The port number to publish on
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    addcsvflags : string, optional
        Specifies the event type to insert into input CSV events.
        Valid values: 'normal' or 'partialupdate'
    addcsvopcode : string, optional
        Prepends an opcode and comma to input CSV events.
    blocksize : int, optional
        Specifies the number of events to include in a published event block
    cefsyslogprefix : string, optional
        When fstype=cef, specifies that CEF events contain the syslog prefix
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    csvfielddelimiter : string, optional
        Specifies the character delimiter for field data in input CSV events
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    growinginputfile : boolean, optional
        Enables reading from a growing input file by publishers
    header : int, optional
        Specifies the number of input lines to skip before starting
        publish operations.
    ignorecsvparseerrors : boolean, optional
        Specifies that when a field in an input CSV event cannot be parsed,
        the event is dropped, an error is logged, and publishing continues.
    maxevents : int, optional
        Specifies the maximum number of events to publish
    noautogenfield : boolean, optional
        Specifies that input events are missing the key field that is
        autogenerated by the source window.
    prebuffer : boolean, optional
        Controls whether event blocks are buffered to an event block vector
        before doing any injects.
    publishwithupsert : boolean, optional
        Build events with opcode=upsert instead of insert
    rate : int, optional
        Specifies the requested transmit rate in events per second
    repeatcount : int, optional
        Specifies the number of times to repeat the publish operation
    transactional : string, optional
        Sets the event block type to transactional.

    Returns
    -------
    :class:`SocketPublisher`

    '''
    connector_key = dict(cls='fs', type='publish', fsname=re.compile(r':(@\w+@|\d+)$'))

    def __init__(self, port=None, fstype=None, name=None, is_active=None,
                 snapshot=None, addcsvflags=None,
                 addcsvopcode=None, blocksize=None, cefsyslogprefix=None,
                 configfilesection=None, csvfielddelimiter=None,
                 dateformat=None,
                 growinginputfile=None, header=None, ignorecsvparseerrors=None,
                 maxevents=None, noautogenfield=None, prebuffer=None,
                 publishwithupsert=None, rate=None, repeatcount=None,
                 transactional=None):
        params = dict(locals())
        params.pop('is_active')
        if isinstance(port, six.string_types):
            port = port.replace(':', '')
        if port is not None:
            params['fsname'] = ':%s' % int(port)
        params.pop('port', None)
        FilePublisher.__init__(**params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        return cls(req[0], name=name, is_active=is_active, **properties)


class SocketSubscriber(FileSubscriber):
    '''
    Subscribe to events from a socket

    Parameters
    ----------
    host : string
        The host name where the socket exists
    port : int
        The port number on the server
    fstype : string, optional
        The data file type.
        Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'
    name : string, optional
        Name of the connector object
    snapshot : boolean, optional
        Specifies whether to send snapshot data
    collapse : boolean, optional
        Converts UPDATE_BLOCK events to UPDATE events in order to make
        subscriber output publishable.
    configfilesection : string, optional
        Specifies the name of the section in the ESP connector config
        file for parameters.
    dateformat : string, optional
        Specifies the format of datetime and timestamp fields
    hdatcashostport : string, optional
        Specifies the CAS server host and port
    hdatcaspassword : string, optional
        Specifies the CAS server password
    hdatcasusername : string, optional
        Specifies the CAS server user name
    hdatfilename : string, optional
        Specifies the name of the Objective Analysis Package Data (HDAT)
        file to be written to the Hadoop Distributed File System (HDFS).
    hdatlasrhostport : string, optional
        Specifies the SAS LASR Analytic Server host and port
    hdatlasrkey : string, optional
        Specifies the path to tklasrkey.sh
    hdatmaxdatanodes : int, optional
        Specifies the maximum number of data node connections
    hdatmaxstringlength : int, optional
        Specifies in bytes the fixed size of string fields in Objective
        Analysis Package Data (HDAT) files
    hdatnumthreads : int, optional
        Specifies the size of the thread pool used for multi-threaded
        writes to data node socket connections.
    hdfsblocksize : int, optional
        Specifies in Mbytes the block size used to write an Objective
        Analysis Package Data (HDAT) file.
    hdfsnumreplicas : int, optional
        Specifies the number of Hadoop Distributed File System (HDFS)
        replicas created with writing an Objective Analysis Package
        Data (HDAT) file.
    header : boolean or string, optional
        For a CSV subscriber, specifies to write a header row that
        shows comma-separated fields.
        Valid values: True, False, or 'full' (include opcode flags in header)
    maxfilesize : int, optional
        Specifies the maximum size in bytes of the subscriber output file
    periodicity : int, optional
        Specifies the interval in seconds at which the subscriber output
        file is closed and a new output file opened.
    rate : boolean, optional
        When latency mode is enabled, shows this specified rate in generated
        output files.
    rmretdel : boolean, optional
        Specifies to remove all delete events from event blocks received
        by a subscriber that were introduced by a window retention policy.
    unbufferedoutputstreams : boolean, optional
        Specifies to create an unbuffered stream when writing to a
        file or socket.

    Returns
    -------
    :class:`SocketSubscriber`

    '''
    connector_key = dict(cls='fs', type='subscribe',
                         fsname=re.compile(r'^(@\w+@|[\w+\.\-]+):(@\w+@|\d+)$'))

    def __init__(self, host=None, port=None, fstype=None, name=None, is_active=None,
                 snapshot=None, collapse=None, configfilesection=None,
                 dateformat=None, hdatcashostport=None,
                 hdatcaspassword=None, hdatcasusername=None,
                 hdatfilename=None, hdatlasrhostport=None, hdatlasrkey=None,
                 hdatmaxdatanodes=None, hdatmaxstringlength=None,
                 hdatnumthreads=None, hdfsblocksize=None, hdfsnumreplicas=None,
                 header=None, maxfilesize=None, periodicity=None,
                 rate=None, rmretdel=None, unbufferedoutputstreams=None):
        params = dict(locals())
        params.pop('is_active')
        if host is not None and port is not None:
            params['fsname'] = '%s:%s' % (host, int(port))
        params.pop('host', None)
        params.pop('port', None)
        FileSubscriber.__init__(**params)

    @classmethod
    def from_parameters(cls, conncls, type=None, name=None, is_active=None,
                        properties=None):
        req, properties = map_properties(cls, properties, required='fsname',
                                         delete='type')
        if req[0] is not None:
            host, port = req[0].split(':', 1)
        return cls(host, port, name=name, is_active=is_active, **properties)
#!/usr/bin/env python # encoding: utf-8 # # Copyright SAS Institute # #  Licensed under the Apache License, Version 2.0 (the License); #  you may not use this file except in compliance with the License. #  You may obtain a copy of the License at # #      http://www.apache.org/licenses/LICENSE-2.0 # #  Unless required by applicable law or agreed to in writing, software #  distributed under the License is distributed on an "AS IS" BASIS, #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #  See the License for the specific language governing permissions and #  limitations under the License. # ESP File and Socket Connectors Publish events from a file      Parameters     ----------     fsname : string         The filename or path of the file     fstype : string, optional         The data file type.         Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'     name : string, optional         Name of the connector object     snapshot : boolean, optional         Specifies whether to send snapshot data     addcsvflags : string, optional         Specifies the event type to insert into input CSV events.         Valid values: 'normal' or 'partialupdate'     addcsvopcode : string, optional         Prepends an opcode and comma to input CSV events.     blocksize : int, optional         Specifies the number of events to include in a published event block     cefsyslogprefix : string, optional         When fstype=cef, specifies that CEF events contain the syslog prefix     configfilesection : string, optional         Specifies the name of the section in the ESP connector config         file for parameters.     csvfielddelimiter : string, optional         Specifies the character delimiter for field data in input CSV events     dateformat : string, optional         Specifies the format of datetime and timestamp fields     growinginputfile : boolean, optional         Enables reading from a growing input file by publishers     header : int, optional         Specifies the number of input lines to skip before starting         publish operations.     ignorecsvparseerrors : boolean, optional         Specifies that when a field in an input CSV event cannot be parsed,         the event is dropped, an error is logged, and publishing continues.     maxevents : int, optional         Specifies the maximum number of events to publish     noautogenfield : boolean, optional         Specifies that input events are missing the key field that is         autogenerated by the source window.     prebuffer : boolean, optional         Controls whether event blocks are buffered to an event block vector         before doing any injects.     publishwithupsert : boolean, optional         Build events with opcode=upsert instead of insert     rate : int, optional         Specifies the requested transmit rate in events per second     repeatcount : int, optional         Specifies the number of times to repeat the publish operation     transactional : string, optional         Sets the event block type to transactional.      Returns     -------     :class:`FilePublisher` Subscribe to events from a file      Parameters     ----------     fsname : string         The filename or path of the file     fstype : string, optional         The data file type.         Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'     name : string, optional         Name of the connector object     snapshot : boolean, optional         Specifies whether to send snapshot data     collapse : boolean, optional         Converts UPDATE_BLOCK events to UPDATE events in order to make         subscriber output publishable.     configfilesection : string, optional         Specifies the name of the section in the ESP connector config         file for parameters.     dateformat : string, optional         Specifies the format of datetime and timestamp fields     hdatcashostport : string, optional         Specifies the CAS server host and port     hdatcaspassword : string, optional         Specifies the CAS server password     hdatcasusername : string, optional         Specifies the CAS server user name     hdatfilename : string, optional         Specifies the name of the Objective Analysis Package Data (HDAT)         file to be written to the Hadoop Distributed File System (HDFS).     hdatlasrhostport : string, optional         Specifies the SAS LASR Analytic Server host and port     hdatlasrkey : string, optional         Specifies the path to tklasrkey.sh     hdatmaxdatanodes : int, optional         Specifies the maximum number of data node connections     hdatmaxstringlength : int, optional         Specifies in bytes the fixed size of string fields in Objective         Analysis Package Data (HDAT) files     hdatnumthreads : int, optional         Specifies the size of the thread pool used for multi-threaded         writes to data node socket connections.     hdfsblocksize : int, optional         Specifies in Mbytes the block size used to write an Objective         Analysis Package Data (HDAT) file.     hdfsnumreplicas : int, optional         Specifies the number of Hadoop Distributed File System (HDFS)         replicas created with writing an Objective Analysis Package         Data (HDAT) file.     header : boolean or string, optional         For a CSV subscriber, specifies to write a header row that         shows comma-separated fields.         Valid values: True, False, or 'full' (include opcode flags in header)     maxfilesize : int, optional         Specifies the maximum size in bytes of the subscriber output file     periodicity : int, optional         Specifies the interval in seconds at which the subscriber output         file is closed and a new output file opened.     rate : boolean, optional         When latency mode is enabled, shows this specified rate in generated         output files.     rmretdel : boolean, optional         Specifies to remove all delete events from event blocks received         by a subscriber that were introduced by a window retention policy.     unbufferedoutputstreams : boolean, optional         Specifies to create an unbuffered stream when writing to a         file or socket.      Returns     -------     :class:`FileSubscriber` Publish events from a socket      Parameters     ----------     port : int         The port number to publish on     fstype : string, optional         The data file type.         Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'     name : string, optional         Name of the connector object     snapshot : boolean, optional         Specifies whether to send snapshot data     addcsvflags : string, optional         Specifies the event type to insert into input CSV events.         Valid values: 'normal' or 'partialupdate'     addcsvopcode : string, optional         Prepends an opcode and comma to input CSV events.     blocksize : int, optional         Specifies the number of events to include in a published event block     cefsyslogprefix : string, optional         When fstype=cef, specifies that CEF events contain the syslog prefix     configfilesection : string, optional         Specifies the name of the section in the ESP connector config         file for parameters.     csvfielddelimiter : string, optional         Specifies the character delimiter for field data in input CSV events     dateformat : string, optional         Specifies the format of datetime and timestamp fields     growinginputfile : boolean, optional         Enables reading from a growing input file by publishers     header : int, optional         Specifies the number of input lines to skip before starting         publish operations.     ignorecsvparseerrors : boolean, optional         Specifies that when a field in an input CSV event cannot be parsed,         the event is dropped, an error is logged, and publishing continues.     maxevents : int, optional         Specifies the maximum number of events to publish     noautogenfield : boolean, optional         Specifies that input events are missing the key field that is         autogenerated by the source window.     prebuffer : boolean, optional         Controls whether event blocks are buffered to an event block vector         before doing any injects.     publishwithupsert : boolean, optional         Build events with opcode=upsert instead of insert     rate : int, optional         Specifies the requested transmit rate in events per second     repeatcount : int, optional         Specifies the number of times to repeat the publish operation     transactional : string, optional         Sets the event block type to transactional.      Returns     -------     :class:`SocketPublisher` Subscribe to events from a socket      Parameters     ----------     host : string         The host name where the socket exists     port : int         The port number on the server     fstype : string, optional         The data file type.         Valid values: 'binary', 'csv', 'xml', 'json', 'syslog', 'hdat', 'cef'     name : string, optional         Name of the connector object     snapshot : boolean, optional         Specifies whether to send snapshot data     collapse : boolean, optional         Converts UPDATE_BLOCK events to UPDATE events in order to make         subscriber output publishable.     configfilesection : string, optional         Specifies the name of the section in the ESP connector config         file for parameters.     dateformat : string, optional         Specifies the format of datetime and timestamp fields     hdatcashostport : string, optional         Specifies the CAS server host and port     hdatcaspassword : string, optional         Specifies the CAS server password     hdatcasusername : string, optional         Specifies the CAS server user name     hdatfilename : string, optional         Specifies the name of the Objective Analysis Package Data (HDAT)         file to be written to the Hadoop Distributed File System (HDFS).     hdatlasrhostport : string, optional         Specifies the SAS LASR Analytic Server host and port     hdatlasrkey : string, optional         Specifies the path to tklasrkey.sh     hdatmaxdatanodes : int, optional         Specifies the maximum number of data node connections     hdatmaxstringlength : int, optional         Specifies in bytes the fixed size of string fields in Objective         Analysis Package Data (HDAT) files     hdatnumthreads : int, optional         Specifies the size of the thread pool used for multi-threaded         writes to data node socket connections.     hdfsblocksize : int, optional         Specifies in Mbytes the block size used to write an Objective         Analysis Package Data (HDAT) file.     hdfsnumreplicas : int, optional         Specifies the number of Hadoop Distributed File System (HDFS)         replicas created with writing an Objective Analysis Package         Data (HDAT) file.     header : boolean or string, optional         For a CSV subscriber, specifies to write a header row that         shows comma-separated fields.         Valid values: True, False, or 'full' (include opcode flags in header)     maxfilesize : int, optional         Specifies the maximum size in bytes of the subscriber output file     periodicity : int, optional         Specifies the interval in seconds at which the subscriber output         file is closed and a new output file opened.     rate : boolean, optional         When latency mode is enabled, shows this specified rate in generated         output files.     rmretdel : boolean, optional         Specifies to remove all delete events from event blocks received         by a subscriber that were introduced by a window retention policy.     unbufferedoutputstreams : boolean, optional         Specifies to create an unbuffered stream when writing to a         file or socket.      Returns     -------     :class:`SocketSubscriber`