#include <uv.h>
char * sirinet_pipe_name(uv_pipe_t * client);
+void sirinet_pipe_unlink(uv_pipe_t * client);
#endif /* SIRINET_PIPE_H_ */
/* make sure data is set to NULL so we later on can check this value. */
client_server_tcp.data = NULL;
+ client_server_pipe.data = NULL;
if (siri->cfg->bind_client_addr != NULL)
{
}
rc = uv_listen(
- (uv_stream_t*) &client_server_tcp,
+ (uv_stream_t *) &client_server_tcp,
DEFAULT_BACKLOG,
on_tcp_new_connection);
if (siri->cfg->pipe_support)
{
- char *pipe_name = siri->cfg->pipe_client_name;
+ char * pipe_name = siri->cfg->pipe_client_name;
- rc = uv_pipe_bind(
- &client_server_pipe,
- pipe_name);
+ rc = uv_pipe_bind(&client_server_pipe, pipe_name);
if (rc)
{
}
rc = uv_listen(
- (uv_stream_t*) &client_server_pipe,
+ (uv_stream_t *) &client_server_pipe,
DEFAULT_BACKLOG,
on_pipe_new_connection);
if (rc)
{
- log_error("Error listening TCP client server: %s", uv_strerror(rc));
+ log_error(
+ "Error listening pipe client server: %s", uv_strerror(rc));
return 1;
}
#include <string.h>
#include <stdlib.h>
#include <siri/net/pipe.h>
+#include <siri/siri.h>
#include <xpath/xpath.h>
#define PIPE_NAME_BUF_SZ SIRI_PATH_MAX
return buffer;
}
+
+void sirinet_pipe_unlink(uv_pipe_t * client)
+{
+ char * pipe_name = sirinet_pipe_name(client);
+ if (pipe_name != NULL)
+ {
+ uv_fs_t * req = malloc(sizeof(uv_fs_t));
+ if (req != NULL)
+ {
+ uv_fs_unlink(siri.loop, req, pipe_name, (uv_fs_cb) free);
+ }
+ free(pipe_name);
+ }
+}
#include <siri/siri.h>
#include <stdlib.h>
#include <string.h>
-#include <siri/siri.h>
#define MAX_ALLOWED_PKG_SIZE 20971520 /* 20 MB */
siri.client = NULL;
break;
case STREAM_PIPE_CLIENT:
+ if (client->origin != NULL)
{
- char * pipe_name = sirinet_pipe_name((uv_pipe_t *) uvclient);
- if (pipe_name != NULL)
- {
- uv_fs_t * req = malloc(sizeof(uv_fs_t));
- if (req != NULL)
- {
- uv_fs_unlink(siri.loop, req, pipe_name, (uv_fs_cb) free);
- }
- free(pipe_name);
- }
+ siridb_user_t * user = client->origin;
+ siridb_user_decref(user);
}
-
+ sirinet_pipe_unlink((uv_pipe_t *) uvclient);
+ break;
}
free(client->buf);
free(client);
--- /dev/null
+#include <uv.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+void OnConnect(uv_connect_t * connect, int status)
+{
+ printf("Hi!");
+}
+
+int main()
+{
+ uv_loop_t * loop = uv_default_loop();
+ uv_pipe_t * handle = malloc(sizeof(uv_pipe_t));
+ uv_connect_t * connect = malloc(sizeof(uv_connect_t));
+
+ uv_pipe_init(loop, handle, 0);
+ uv_pipe_open(handle, socket(PF_UNIX, SOCK_STREAM, 0));
+ uv_pipe_connect(connect, handle, "/tmp/siridb_pipe_test.sock", OnConnect);
+
+ uv_run(loop, UV_RUN_DEFAULT);
+}
\ No newline at end of file
--- /dev/null
+import os
+import asyncio
+import functools
+import random
+import time
+from testing import Client
+from testing import default_test_setup
+from testing import gen_data
+from testing import gen_points
+from testing import gen_series
+from testing import InsertError
+from testing import PoolError
+from testing import QueryError
+from testing import run_test
+from testing import Series
+from testing import Server
+from testing import ServerError
+from testing import SiriDB
+from testing import TestBase
+from testing import UserAuthError
+from testing import SiriDBAsyncUnixConnection
+
+PIPE_NAME = '/tmp/siridb_pipe_test.sock'
+
+DATA = {
+ 'series num_float': [
+ [1471254705, 1.5],
+ [1471254707, -3.5],
+ [1471254710, -7.3]],
+ 'series num_integer': [
+ [1471254705, 5],
+ [1471254708, -3],
+ [1471254710, -7]],
+ 'series_log': [
+ [1471254710, 'log line one'],
+ [1471254712, 'log line two'],
+ [1471254714, 'another line (three)'],
+ [1471254716, 'and yet one more']]
+}
+
+if os.path.exists(PIPE_NAME):
+ os.unlink(PIPE_NAME)
+
+class TestPipeSupport(TestBase):
+ title = 'Test pipe support object'
+
+
+ @default_test_setup(1, pipe_name=PIPE_NAME)
+ async def run(self):
+
+ pipe_client = SiriDBAsyncUnixConnection(PIPE_NAME)
+
+ await pipe_client.connect('iris', 'siri', self.db.dbname)
+
+ await pipe_client.insert(DATA)
+
+ self.assertEqual(
+ await pipe_client.query('select * from "series_log"'),
+ {'series_log': DATA['series_log']})
+
+ pipe_client.close()
+
+
+ # return False
+
+
+if __name__ == '__main__':
+ SiriDB.LOG_LEVEL = 'CRITICAL'
+ Server.HOLD_TERM = True
+ Server.MEM_CHECK = True
+ Server.BUILDTYPE = 'Debug'
+ run_test(TestPipeSupport())
from .testbase import default_test_setup
from .testbase import TestBase
from .series import Series
+from .pipe_client import PipeClient as SiriDBAsyncUnixConnection
async def _run_test(test, loglevel):
BUILDTYPE = 'Debug'
TEST_DIR = './testdir'
SIRIDBC = '../{BUILDTYPE}/siridb-server'
-MANAGE = '' # '/usr/sbin/siridb-manage'
ADMIN = '/usr/local/bin/siridb-admin'
-TOOL = 'ADMIN' # TOOL should be ADMIN or MANAGE. (MANAGE is the old, obsolete tool and ADMIN is preffered)
# VALGRIND = 'valgrind --tool=memcheck --leak-check=full --show-leak-kinds=all --track-origins=yes -v '
VALGRIND = 'valgrind --tool=memcheck '
MAX_OPEN_FILES = 512 # Default value is 32768 but with valgrind 512 is max
--- /dev/null
+import os
+import logging
+import asyncio
+from siridb.connector import SiriDBProtocol
+from siridb.connector.lib.connection import SiriDBAsyncConnection
+
+
+class PipeClient(SiriDBAsyncConnection):
+ def __init__(self, pipe_name, loop=None):
+ self._pipe_name = pipe_name
+ self._protocol = None
+
+ async def connect(self, username, password, dbname, loop=None):
+ loop = loop or asyncio.get_event_loop()
+
+ transport, self._protocol = await loop.create_unix_connection(
+ path=self._pipe_name,
+ protocol_factory=lambda: SiriDBProtocol(
+ username, password, dbname))
+
+ try:
+ res = await self._protocol.auth_future
+ except Exception as exc:
+ logging.debug('Authentication failed: {}'.format(exc))
+ transport.close()
+ raise exc
+ else:
+ self._protocol.on_authenticated()
+
+
optimize_interval=30,
heartbeat_interval=30,
compression=True,
+ pipe_name=None,
**unused):
self.n = n
self.compression = compression
+ self.enable_pipe_support = int(bool(pipe_name))
+ self.pipe_name = \
+ 'siridb_client.sock' if not self.enable_pipe_support else \
+ pipe_name
self.listen_client_port = 9000 + n
self.listen_backend_port = 9010 + n
self._server_address = self.SERVER_ADDRESS
config.set('siridb', 'default_db_path', self.dbpath)
config.set('siridb', 'max_open_files', MAX_OPEN_FILES)
config.set('siridb', 'enable_shard_compression', int(self.compression))
+ config.set('siridb', 'enable_pipe_support', self.enable_pipe_support)
+ config.set('siridb', 'pipe_client_name', self.pipe_name)
with open(self.cfgfile, 'w') as configfile:
config.write(configfile)
import logging
import random
import asyncio
-from .constants import MANAGE
from .constants import ADMIN
-from .constants import TOOL
VERBOSE = ' --verbose'
self.dbname,
server.name))
- if TOOL == 'MANAGE':
- rc = os.system(
- '{manage} '
- '--log-level {log_level} '
- '--noroot --config {cfgfile} create-new '
- '--dbname {dbname} '
- '--time-precision {time_precision} '
- '{bufpath}'
- '--duration-log {duration_log} '
- '--duration-num {duration_num} '
- '--buffer-size {buffer_size}'.format(
- manage=MANAGE,
- log_level=self.LOG_LEVEL.lower(),
- cfgfile=server.cfgfile,
- bufpath='' if not self.buffer_path
- else '--buffer-path {}'.format(self.buffer_path),
- **vars(self)))
- elif TOOL == 'ADMIN':
- rc = os.system(
- '{admin} '
- '-u sa -p siri -s {addr} '
- 'new-database '
- '--db-name {dbname} '
- '--time-precision {time_precision} '
- '--duration-log {duration_log} '
- '--duration-num {duration_num} '
- '--buffer-size {buffer_size}'
- '{verbose}'.format(
- admin=ADMIN,
- addr=server.addr,
- verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
- else ' >/dev/null',
- **vars(self)))
- else:
- logging.error(
- 'TOOL should be either MANAGE or ADMIN, got: {}'
- .format(TOOL))
- rc = 1
+ rc = os.system(
+ '{admin} '
+ '-u sa -p siri -s {addr} '
+ 'new-database '
+ '--db-name {dbname} '
+ '--time-precision {time_precision} '
+ '--duration-log {duration_log} '
+ '--duration-num {duration_num} '
+ '--buffer-size {buffer_size}'
+ '{verbose}'.format(
+ admin=ADMIN,
+ addr=server.addr,
+ verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
+ else ' >/dev/null',
+ **vars(self)))
assert rc == 0, 'Expected rc = 0 but got rc = {}'.format(rc)
if remote_server is None:
remote_server = random.choice(self.servers)
- if TOOL == 'MANAGE':
- rc = os.system(
- '{manage} '
- '--log-level {log_level} '
- '--noroot --config {cfgfile} create-replica '
- '--dbname {dbname} '
- '--remote-address {remote_address} '
- '--remote-port {remote_port} '
- '--user {user} '
- '--password {password} '
- '--pool {pool} '
- '{bufpath}'
- '--buffer-size {buffer_size}'.format(
- manage=MANAGE,
- log_level=self.LOG_LEVEL.lower(),
- cfgfile=server.cfgfile,
- user=username,
- password=password,
- pool=pool,
- bufpath='' if not self.buffer_path
- else '--buffer-path {}'.format(self.buffer_path),
- **vars(self),
- remote_address=remote_server.server_address,
- remote_port=remote_server.listen_client_port))
- elif TOOL == 'ADMIN':
- rc = os.system(
- '{admin} '
- '-u sa -p siri -s {addr} '
- 'new-replica '
- '--db-name {dbname} '
- '--db-server {dbaddr} '
- '--db-user {dbuser} '
- '--db-password {dbpassword} '
- '--pool {pool} '
- '--force{verbose}'.format(
- admin=ADMIN,
- addr=server.addr,
- dbaddr=remote_server.addr,
- dbuser=username,
- dbpassword=password,
- pool=pool,
- verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
- else ' >/dev/null',
- **vars(self)))
- else:
- logging.error(
- 'TOOL should be either MANAGE or ADMIN, got: {}'
- .format(TOOL))
- rc = 1
+ rc = os.system(
+ '{admin} '
+ '-u sa -p siri -s {addr} '
+ 'new-replica '
+ '--db-name {dbname} '
+ '--db-server {dbaddr} '
+ '--db-user {dbuser} '
+ '--db-password {dbpassword} '
+ '--pool {pool} '
+ '--force{verbose}'.format(
+ admin=ADMIN,
+ addr=server.addr,
+ dbaddr=remote_server.addr,
+ dbuser=username,
+ dbpassword=password,
+ pool=pool,
+ verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
+ else ' >/dev/null',
+ **vars(self)))
assert rc == 0, 'Expected rc = 0 but got rc = {}'.format(rc)
if remote_server is None:
remote_server = random.choice(self.servers)
- if TOOL == 'MANAGE':
- rc = os.system(
- '{manage} '
- '--log-level {log_level} '
- '--noroot --config {cfgfile} create-pool '
- '--dbname {dbname} '
- '--remote-address {remote_address} '
- '--remote-port {remote_port} '
- '--user {user} '
- '--password {password} '
- '{bufpath}'
- '--buffer-size {buffer_size}'.format(
- manage=MANAGE,
- log_level=self.LOG_LEVEL.lower(),
- cfgfile=server.cfgfile,
- user=username,
- password=password,
- bufpath='' if not self.buffer_path
- else '--buffer-path {}'.format(self.buffer_path),
- **vars(self),
- remote_address=remote_server.server_address,
- remote_port=remote_server.listen_client_port))
- elif TOOL == 'ADMIN':
- rc = os.system(
- '{admin} '
- '-u sa -p siri -s {addr} '
- 'new-pool '
- '--db-name {dbname} '
- '--db-server {dbaddr} '
- '--db-user {dbuser} '
- '--db-password {dbpassword} '
- '--force{verbose}'.format(
- admin=ADMIN,
- addr=server.addr,
- dbaddr=remote_server.addr,
- dbuser=username,
- dbpassword=password,
- verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
- else ' >/dev/null',
- **vars(self)))
- else:
- logging.error(
- 'TOOL should be either MANAGE or ADMIN, got: {}'
- .format(TOOL))
- rc = 1
+ rc = os.system(
+ '{admin} '
+ '-u sa -p siri -s {addr} '
+ 'new-pool '
+ '--db-name {dbname} '
+ '--db-server {dbaddr} '
+ '--db-user {dbuser} '
+ '--db-password {dbpassword} '
+ '--force{verbose}'.format(
+ admin=ADMIN,
+ addr=server.addr,
+ dbaddr=remote_server.addr,
+ dbuser=username,
+ dbpassword=password,
+ verbose=VERBOSE if self.LOG_LEVEL == 'DEBUG'
+ else ' >/dev/null',
+ **vars(self)))
assert rc == 0, 'Expected rc = 0 but got rc = {}'.format(rc)
projects / siridb-server.git / commitdiff