flutter_flutter/tools/skydb

#!/usr/bin/env python
# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from skypy.paths import Paths
from skypy.skyserver import SkyServer
import argparse
import json
import logging
import os
import requests
import signal
import skypy.configuration as configuration
import subprocess
import urlparse
import time


SUPPORTED_MIME_TYPES = [
    'text/html',
    'text/sky',
    'text/plain',
]

DEFAULT_SKY_COMMAND_PORT = 7777
SKY_SERVER_PORT = 9999
PID_FILE_PATH = "/tmp/skydb.pids"
DEFAULT_URL = "https://raw.githubusercontent.com/domokit/mojo/master/sky/examples/home.sky"


class SkyDebugger(object):
    def __init__(self):
        self.paths = None
        self.pids = {}

    def _server_root_for_url(self, url_or_path):
        path = os.path.abspath(url_or_path)
        if os.path.commonprefix([path, self.paths.src_root]) == self.paths.src_root:
            server_root = self.paths.src_root
        else:
            server_root = os.path.dirname(path)
            logging.warn(
                '%s is outside of mojo root, using %s as server root' %
                (path, server_root))
        return server_root

    def _in_chromoting(self):
        return os.environ.get('CHROME_REMOTE_DESKTOP_SESSION', False)

    def _build_mojo_shell_command(self, args):
        self.paths = Paths(os.path.join('out', args.configuration))

        content_handlers = ['%s,%s' % (mime_type, 'mojo:sky_viewer')
            for mime_type in SUPPORTED_MIME_TYPES]
        shell_command = [
            self.paths.mojo_shell_path,
            '--v=1',
            '--content-handlers=%s' % ','.join(content_handlers),
            '--url-mappings=mojo:window_manager=mojo:sky_debugger',
            '--args-for=mojo:sky_debugger_prompt %d' % args.command_port,
            'mojo:window_manager',
        ]
        if args.use_osmesa:
            shell_command.append('--args-for=mojo:native_viewport_service --use-osmesa')

        if args.gdb:
            shell_command = ['gdb', '--args'] + shell_command

        return shell_command

    def start_command(self, args):
        shell_command = self._build_mojo_shell_command(args)
        if args.show_command:
            print " ".join(shell_command)
            return

        self.stop_command(None) # Quit any existing process.

        print args.url_or_path
        # We only start a server for paths:
        if not urlparse.urlparse(args.url_or_path).scheme:
            server_root = self._server_root_for_url(args.url_or_path)
            sky_server = SkyServer(self.paths, SKY_SERVER_PORT,
                args.configuration, server_root)
            self.pids['sky_server_pid'] = sky_server.start()
            self.pids['sky_server_port'] = sky_server.port
            self.pids['sky_server_root'] = sky_server.root

        self.pids['mojo_shell_pid'] = subprocess.Popen(shell_command).pid
        self.pids['sky_command_port'] = args.command_port

        if not self._wait_for_sky_command_port():
            logging.error('Failed to start sky')
            self.stop_command(None)
        else:
            self.load_command(args)

    def _kill_if_exists(self, key, name):
        pid = self.pids.pop(key, None)
        if not pid:
            logging.info('No pid for %s, nothing to do.' % name)
            return
        logging.info('Killing %s (%s).' % (name, pid))
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            logging.info('%s (%s) already gone.' % (name, pid))

    def stop_command(self, args):
        # FIXME: Send /quit to sky prompt instead of killing.
        # self._send_command_to_sky('/quit')
        self._kill_if_exists('mojo_shell_pid', 'mojo_shell')
        self._kill_if_exists('sky_server_pid', 'sky_server')

    def load_command(self, args):
        if not urlparse.urlparse(args.url_or_path).scheme:
            url = SkyServer.url_for_path(self.pids['sky_server_port'],
                    self.pids['sky_server_root'], args.url_or_path)
        else:
            url = args.url_or_path
        self._send_command_to_sky('/load', url)

    def _command_base_url(self):
        return 'http://localhost:%s' % self.pids['sky_command_port']

    def _send_command_to_sky(self, command_path, payload=None):
        url = 'http://localhost:%s%s' % (
            self.pids['sky_command_port'], command_path)
        if payload:
            response = requests.post(url, payload)
        else:
            response = requests.get(url)
        print response.text

    # FIXME: These could be made into a context object with __enter__/__exit__.
    def _load_pid_file(self, path):
        try:
            with open(path, 'r') as pid_file:
                return json.load(pid_file)
        except:
            if os.path.exists(path):
                logging.warn('Failed to read pid file: %s' % path)
            return {}

    def _write_pid_file(self, path, pids):
        try:
            with open(path, 'w') as pid_file:
                json.dump(pids, pid_file)
        except:
            logging.warn('Failed to write pid file: %s' % path)

    def _add_basic_command(self, subparsers, name, url_path, help_text):
        parser = subparsers.add_parser(name, help=help_text)
        command = lambda args: self._send_command_to_sky(url_path)
        parser.set_defaults(func=command)

    def _wait_for_sky_command_port(self):
        tries = 0
        while True:
            try:
                self._send_command_to_sky('/')
                return True
            except:
                tries += 1
                if tries == 3:
                    logging.warn('Still waiting for sky on port %s' %
                        self.pids['sky_command_port'])
                if tries > 10:
                    return False
                time.sleep(1)

    def main(self):
        logging.basicConfig(level=logging.INFO)

        self.pids = self._load_pid_file(PID_FILE_PATH)

        parser = argparse.ArgumentParser(description='Sky launcher/debugger')
        subparsers = parser.add_subparsers(help='sub-command help')

        start_parser = subparsers.add_parser('start',
            help='launch a new mojo_shell with sky')
        configuration.add_arguments(start_parser)
        start_parser.add_argument('--gdb', action='store_true')
        start_parser.add_argument('--command-port', type=int,
            default=DEFAULT_SKY_COMMAND_PORT)
        start_parser.add_argument('--use-osmesa', action='store_true',
            default=self._in_chromoting())
        start_parser.add_argument('url_or_path', nargs='?', type=str,
            default=DEFAULT_URL)
        start_parser.add_argument('--show-command', action='store_true',
            help='Display the shell command and exit')
        start_parser.set_defaults(func=self.start_command)

        stop_parser = subparsers.add_parser('stop',
            help=('stop sky (as listed in %s)' % PID_FILE_PATH))
        stop_parser.set_defaults(func=self.stop_command)

        self._add_basic_command(subparsers, 'trace', '/trace',
            'toggle tracing')
        self._add_basic_command(subparsers, 'reload', '/reload',
            'reload the current page')
        self._add_basic_command(subparsers, 'inspect', '/inspect',
            'stop the running sky instance')

        load_parser = subparsers.add_parser('load',
            help='load a new page in the currently running sky')
        load_parser.add_argument('url_or_path', type=str)
        load_parser.set_defaults(func=self.load_command)

        args = parser.parse_args()
        args.func(args)

        self._write_pid_file(PID_FILE_PATH, self.pids)


if __name__ == '__main__':
    SkyDebugger().main()