File size: 4,314 Bytes
a65138c | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | # -*- coding: utf-8 -*-
#
# pyhwp : hwp file format parser in python
# Copyright (C) 2010 https://github.com/mete0r
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from __future__ import with_statement
import contextlib
import logging
logger = logging.getLogger(__name__)
@contextlib.contextmanager
def soffice_subprocess(**kwargs):
''' Create an remote instance of soffice '''
args = [kwargs.get('soffice', 'soffice')]
if 'accept' in kwargs:
args.append('--accept=%s' % kwargs['accept'])
if kwargs.get('headless', True):
args.append('--headless')
if kwargs.get('invisible', True):
args.append('--invisible')
if kwargs.get('nologo', True):
args.append('--nologo')
if kwargs.get('norestore', True):
args.append('--norestore')
if kwargs.get('nodefault', True):
args.append('--nodefault')
if kwargs.get('nofirstwizard', True):
args.append('--nofirstwizard')
import subprocess
p = subprocess.Popen(args)
pid = p.pid
logger.info('soffice(%s) has been started.', pid)
try:
yield p
finally:
import time
n = 0
p.poll()
while p.returncode is None:
n += 1
if n > 3:
p.kill()
logger.info('trying to kill soffice(%s)', pid)
return
p.terminate()
time.sleep(1)
p.poll()
logger.info('soffice(%s) has been terminated with exit code %d',
pid, p.returncode)
def connect_remote_context(uno_link, max_tries=10):
''' Connect to the remote soffice instance and get the context. '''
from unokit.services import css
resolver = css.bridge.UnoUrlResolver()
uno_url = 'uno:'+uno_link+'StarOffice.ComponentContext'
logger.info('uno_url: %s', uno_url)
from com.sun.star.connection import NoConnectException
while True:
max_tries -= 1
try:
return resolver.resolve(uno_url)
except NoConnectException, e:
if max_tries <= 0:
raise
logger.info('%s - retrying', type(e).__name__)
import time
time.sleep(1)
continue
@contextlib.contextmanager
def new_remote_context(pipe='oxt.tool', retry=3, make_current=True, **kwargs):
''' Create a remote soffice instance and get its context
:param pipe: connection pipe name
:param retry: connect retry count; default True.
:param make_current: whether the remote context would be pushed to be
current context; default True.
:param **kwargs: arguments to soffice_subprocess()
:returns: remote context
'''
uno_link = 'pipe,name=%s;urp;' % pipe
logger.debug('uno_link: %s', uno_link)
kwargs['accept'] = uno_link
while retry >= 0:
with soffice_subprocess(**kwargs):
import time
time.sleep(1)
try:
context = connect_remote_context(uno_link, max_tries=10)
except Exception, e:
logger.exception(e)
retry -= 1
continue
if make_current:
import unokit.contexts
unokit.contexts.push(context)
try:
yield context
finally:
unokit.contexts.pop()
else:
yield context
return
class RemoteContextLayer:
@classmethod
def setUp(cls):
cls.context = new_remote_context()
cls.context.__enter__()
@classmethod
def tearDown(cls):
cls.context.__exit__(None, None, None)
|