text:
// ==========================================================================
//                      Yara - Yet Another Read Aligner
// ==========================================================================
// Copyright (c) 2011-2014, Enrico Siragusa, FU Berlin
// Copyright (c) 2013 NVIDIA Corporation
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//     * Redistributions of source code must retain the above copyright
//       notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of Enrico Siragusa or the FU Berlin nor the names of
//       its contributors may be used to endorse or promote products derived
//       from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL ENRICO SIRAGUSA OR THE FU BERLIN BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
// LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
// OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
// DAMAGE.
//
// ==========================================================================
// Author: Enrico Siragusa <enrico.siragusa@fu-berlin.de>
// ==========================================================================
// This file contains Index overloads.
// ==========================================================================

#ifndef APP_YARA_INDEX_FM_H_
#define APP_YARA_INDEX_FM_H_

namespace seqan {

// ============================================================================
// Functions
// ============================================================================

// ----------------------------------------------------------------------------
// Function indexRequire()
// ----------------------------------------------------------------------------
// This function is overloaded to avoid compiling the code building the index.

#ifndef YARA_INDEXER
template <typename TText, typename TSSetSpec, typename TSpec, typename TConfig, typename TFibre>
inline bool indexRequire(Index<StringSet<TText, TSSetSpec>,  FMIndex<TSpec, TConfig> > & index, Tag<TFibre> const fibre)
{
    if (!indexSupplied(index, fibre))
        throw RuntimeError("The reference index file was not loaded correctly.");

    return true;
}
#endif

// ----------------------------------------------------------------------------
// Function save()
// ----------------------------------------------------------------------------
// This function is overloaded to avoid saving the text.

template <typename TText, typename TSSetSpec, typename TSpec, typename TConfig>
inline bool save(Index<StringSet<TText, TSSetSpec>, FMIndex<TSpec, TConfig> > const & index,
                 const char * fileName, int openMode)
{
    String<char> name;

    name = fileName;    append(name, ".sa");
    if (!save(getFibre(index, FibreSA()), toCString(name), openMode)) return false;

    name = fileName;    append(name, ".lf");
    if (!save(getFibre(index, FibreLF()), toCString(name), openMode)) return false;

    return true;
}

// ----------------------------------------------------------------------------
// Function open()
// ----------------------------------------------------------------------------
// This function is overloaded to avoid loading the text.

template <typename TText, typename TSSetSpec, typename TSpec, typename TConfig>
inline bool open(Index<StringSet<TText, TSSetSpec>, FMIndex<TSpec, TConfig> > & index,
                 const char * fileName, int openMode)
{
    String<char> name;

    name = fileName;    append(name, ".sa");
    if (!open(getFibre(index, FibreSA()), toCString(name), openMode)) return false;

    name = fileName;    append(name, ".lf");
    if (!open(getFibre(index, FibreLF()), toCString(name), openMode)) return false;

    setFibre(getFibre(index, FibreSA()), getFibre(index, FibreLF()), FibreLF());

    return true;
}

// ----------------------------------------------------------------------------
// Function _getNodeByChar()
// ----------------------------------------------------------------------------
// This function is overloaded for Dna5 and Dna5Q to let Ns always mismatch.

template <typename TText, typename TOccSpec, typename TSize_, typename TLen_, typename TSum_, typename TAlloc_, typename TSpec, typename TSize>
inline bool
_getNodeByChar(Iter<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >, VSTree<TopDown<TSpec> > > const & it,
               typename VertexDescriptor<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type const & vDesc,
               Pair<typename Size<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type> & _range,
               TSize & smaller,
               Dna5 c)
{
    return _getNodeByCharImpl(it, vDesc, _range, smaller, c);
}

template <typename TText, typename TOccSpec, typename TSize_, typename TLen_, typename TSum_, typename TAlloc_, typename TSpec, typename TSize>
inline bool
_getNodeByChar(Iter<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >, VSTree<TopDown<TSpec> > > const & it,
               typename VertexDescriptor<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type const & vDesc,
               Pair<typename Size<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type> & _range,
               TSize & smaller,
               Dna5Q c)
{
    return _getNodeByCharImpl(it, vDesc, _range, smaller, c);
}

template <typename TText, typename TOccSpec, typename TSize_, typename TLen_, typename TSum_, typename TAlloc_, typename TSpec, typename TSize, typename TChar>
inline bool
_getNodeByCharImpl(Iter<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >, VSTree<TopDown<TSpec> > > const & it,
               typename VertexDescriptor<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type const & vDesc,
               Pair<typename Size<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type> & _range,
               TSize & /*smaller*/,
               TChar c)
{
    typedef Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >        TIndex;
    typedef typename Fibre<TIndex, FibreLF>::Type               TLF;
    typedef typename Value<TIndex>::Type                        TAlphabet;

    TIndex const & index = container(it);
    TLF const & lf = indexLF(index);

    if (ordValue(c) >= ValueSize<TAlphabet>::VALUE) return false;

    _range = range(index, vDesc);

    _range.i1 = lf(_range.i1, c);
    _range.i2 = lf(_range.i2, c);

    return _range.i1 < _range.i2;
}

// This function is overloaded for YaraFMConfig because the LevelDictionary for the occurrence table
// does not support cumulativeSearch (calculating the smaller value) yet.
template <typename TText, typename TOccSpec, typename TSize_, typename TLen_, typename TSum_, typename TAlloc_, typename TSpec, typename TSize, typename TChar>
inline bool
_getNodeByChar(Iter<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >, VSTree<TopDown<TSpec> > > const & it,
               typename VertexDescriptor<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type const & vDesc,
               Pair<typename Size<Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > > >::Type> & _range,
               TSize & /*smaller*/,
               TChar c)
{
    typedef Index<TText, FMIndex<TOccSpec, YaraFMConfig<TSize_, TLen_, TSum_, TAlloc_> > >        TIndex;
    typedef typename Fibre<TIndex, FibreLF>::Type               TLF;

    TIndex const & index = container(it);
    TLF const & lf = indexLF(index);

    _range = range(index, vDesc);

    _range.i1 = lf(_range.i1, c);
    _range.i2 = lf(_range.i2, c);

    return _range.i1 < _range.i2;
}

}

#endif  // #ifndef APP_YARA_INDEX_FM_H_


text:
const React = require('react');

module.exports = function(props){
  if (!props.url) return '';
  return (

          Book a table

  );
};

redpajama_set_name: RedPajamaGithub


text:
<?php

/* BlogBundle:Public:menu.html.twig */
class __TwigTemplate_484521475bda767426314119ecc0f2a3f7dfe113b8a230aed644ee7893d5b9dc extends Twig_Template
{
    public function __construct(Twig_Environment $env)
    {
        parent::__construct($env);

        $this->parent = false;

        $this->blocks = array(
        );
    }

    protected function doDisplay(array $context, array $blocks = array())
    {
        // line 1
        echo "
\t

\t\t

\t\t\t
\t\t\t\tenv, $this->env->getExtension('assets')->getAssetUrl("img/logo.png"), "html", null, true);
        echo "\" alt=\"Logo\" width=\"352\" height=\"140\">
\t\t\t
\t\t

\t\t

\t\t\t<button id=\"hamburger\" type=\"button\" class=\"navbar-toggle\" data-toggle=\"collapse\" data-target=\"#readable-navbar-collapse\">
\t\t\t\tToggle navigation
\t\t\t\t
\t\t\t\t
\t\t\t\t
\t\t\t</button>
\t\t

\t\t
\t\t\t

\t\t\t\t

\t\t\t\t\t

\t\t\t\t\t\tenv->getExtension('routing')->getPath("blog_homepage");
        echo "\" class=\"dropdown-toggle\" data-toggle=\"dropdown\">Accueil

\t\t\t\t\t

\t\t\t\t\t";
        // line 22
        $context['_parent'] = (array) $context;
        $context['_seq'] = twig_ensure_traversable((isset($context["categories"]) ? $context["categories"] : $this->getContext($context, "categories")));
        foreach ($context['_seq'] as $context["_key"] => $context["categorie"]) {
            // line 23
            echo "\t\t\t\t\t\t

\t\t\t\t\t\t\tenv, $this->env->getExtension('routing')->getPath("page_categorie", array("id" => $this->getAttribute($context["categorie"], "id", array()), "name" => $this->getAttribute($context["categorie"], "name", array()))), "html", null, true);
            echo "\" class=\"dropdown-toggle\" data-toggle=\"dropdown\">";
            echo twig_escape_filter($this->env, $this->getAttribute($context["categorie"], "name", array()), "html", null, true);
            echo "

\t\t\t\t\t\t

\t\t\t\t\t";
        }
        $_parent = $context['_parent'];
        unset($context['_seq'], $context['_iterated'], $context['_key'], $context['categorie'], $context['_parent'], $context['loop']);
        $context = array_intersect_key($context, $_parent) + $_parent;
        // line 27
        echo "\t\t\t\t

\t\t\t

\t\t
\t

";
    }

    public function getTemplateName()
    {
        return "BlogBundle:Public:menu.html.twig";
    }

    public function isTraitable()
    {
        return false;
    }

    public function getDebugInfo()
    {
        return array (  66 => 27,  55 => 24,  52 => 23,  48 => 22,  43 => 20,  25 => 5,  19 => 1,);
    }
}


text:
package org.apache.geode.internal.cache.region.entry;

import org.apache.geode.internal.cache.*;

public class RegionEntryFactoryBuilder {
  public RegionEntryFactory getRegionEntryFactoryOrNull(boolean statsEnabled, boolean isLRU,
      boolean isDisk, boolean withVersioning, boolean offHeap) {
    int bitRepresentation = 0;
    bitRepresentation |= statsEnabled ? 1 : 0;
    bitRepresentation |= isLRU ? 2 : 0;
    bitRepresentation |= isDisk ? 4 : 0;
    bitRepresentation |= withVersioning ? 8 : 0;
    bitRepresentation |= offHeap ? 16 : 0;

    /**
     * The bits represent all options |offHeap|versioning|disk|lru|stats|
     */
    switch (bitRepresentation) {
      case (0):
        return VMThinRegionEntryHeap.getEntryFactory(); // Bits: 00000
      case (1):
        return VMStatsRegionEntryHeap.getEntryFactory(); // Bits: 00001
      case (2):
        return VMThinLRURegionEntryHeap.getEntryFactory(); // Bits: 00010
      case (3):
        return VMStatsLRURegionEntryHeap.getEntryFactory(); // Bits: 00011
      case (4):
        return VMThinDiskRegionEntryHeap.getEntryFactory(); // Bits: 00100
      case (5):
        return VMStatsDiskRegionEntryHeap.getEntryFactory(); // Bits: 00101
      case (6):
        return VMThinDiskLRURegionEntryHeap.getEntryFactory(); // Bits: 00110
      case (7):
        return VMStatsDiskLRURegionEntryHeap.getEntryFactory(); // Bits: 00111
      case (8):
        return VersionedThinRegionEntryHeap.getEntryFactory(); // Bits: 01000
      case (9):
        return VersionedStatsRegionEntryHeap.getEntryFactory(); // Bits: 01001
      case (10):
        return VersionedThinLRURegionEntryHeap.getEntryFactory(); // Bits: 01010
      case (11):
        return VersionedStatsLRURegionEntryHeap.getEntryFactory(); // Bits: 01011
      case (12):
        return VersionedThinDiskRegionEntryHeap.getEntryFactory(); // Bits: 01100
      case (13):
        return VersionedStatsDiskRegionEntryHeap.getEntryFactory(); // Bits: 01101
      case (14):
        return VersionedThinDiskLRURegionEntryHeap.getEntryFactory(); // Bits: 01110
      case (15):
        return VersionedStatsDiskLRURegionEntryHeap.getEntryFactory(); // Bits: 01111
      case (16):
        return VMThinRegionEntryOffHeap.getEntryFactory(); // Bits: 10000
      case (17):
        return VMStatsRegionEntryOffHeap.getEntryFactory(); // Bits: 10001
      case (18):
        return VMThinLRURegionEntryOffHeap.getEntryFactory(); // Bits: 10010
      case (19):
        return VMStatsLRURegionEntryOffHeap.getEntryFactory(); // Bits: 10011
      case (20):
        return VMThinDiskRegionEntryOffHeap.getEntryFactory(); // Bits: 10100
      case (21):
        return VMStatsDiskRegionEntryOffHeap.getEntryFactory(); // Bits: 10101
      case (22):
        return VMThinDiskLRURegionEntryOffHeap.getEntryFactory(); // Bits: 10110
      case (23):
        return VMStatsDiskLRURegionEntryOffHeap.getEntryFactory(); // Bits: 10111
      case (24):
        return VersionedThinRegionEntryOffHeap.getEntryFactory(); // Bits: 11000
      case (25):
        return VersionedStatsRegionEntryOffHeap.getEntryFactory(); // Bits: 11001
      case (26):
        return VersionedThinLRURegionEntryOffHeap.getEntryFactory(); // Bits: 11010
      case (27):
        return VersionedStatsLRURegionEntryOffHeap.getEntryFactory(); // Bits: 11011
      case (28):
        return VersionedThinDiskRegionEntryOffHeap.getEntryFactory(); // Bits: 11100
      case (29):
        return VersionedStatsDiskRegionEntryOffHeap.getEntryFactory(); // Bits: 11101
      case (30):
        return VersionedThinDiskLRURegionEntryOffHeap.getEntryFactory(); // Bits: 11110
      case (31):
        return VersionedStatsDiskLRURegionEntryOffHeap.getEntryFactory(); // Bits: 11111
      default:
        return null;
    }
  }
}


text:
#region BSD License

#endregion

namespace Clide
{
    using System.Windows;

    /// <summary>
	/// Interaction logic for InputBox.xaml
	/// </summary>
	public partial class InputBox : Window
	{
        /// <summary>
        /// Initializes a new instance of the <see cref="InputBox"/> class.
        /// </summary>
		public InputBox()
		{
			InitializeComponent();
		}

        /// <summary>
        /// Gets or sets the response text.
        /// </summary>
		public string ResponseText
		{
			get { return this.ResponseTextBox.Text; }
			set { this.ResponseTextBox.Text = value; }
		}

        /// <summary>
        /// Gets or sets the message.
        /// </summary>
		public string Message
		{
			get { return this.MessageText.Text; }
			set { this.MessageText.Text = value; }
		}

        /// <summary>
        /// Shows the input box to the user with the specified message.
        /// </summary>
        /// <param name="message">The message to show.</param>
        /// <param name="owner">The window owner.</param>
        /// <returns></returns>
		public static string Show(string message, Window owner = null)
		{
			var dialog = new InputBox() { Message = message };
			dialog.Owner = owner ?? Application.Current.MainWindow;
			if (dialog.ShowDialog() == true)
			{
				return dialog.ResponseText;
			}

			return null;
		}

		private void OkButton_Click(object sender, RoutedEventArgs e)
		{
			this.DialogResult = true;
		}
	}
}


text:
/*
 * Filter that only returns cases.  Used by the change listener.
 */
function(doc, req)
{
    return (doc["doc_type"] == "CommCareCase" || doc["doc_type"] == "CommCareCase-Deleted");
}


text:
<?xml version="1.0"?>
<?xml-stylesheet href="doc.xsl"?>

<doc>Hello</doc>


text:
"""Encapsulates the ibapipy.client_socket.ClientSocket class and provides an
[ideally] simpler API for interacting with TWS.

"""
from asyncio import Future
import asyncio
import logging
import ibapipy.data.contract as ibc
import ibapipy.data.execution_filter as ibef
import ibclientpy.client_adapter as ibca
import ibclientpy.historical_data as ibhd
import ibclientpy.order_handler as iboh
import ibclientpy.config as config

LOG = logging.getLogger(__name__)

class Client:
    """Simplified interface to an ibapipy.client_socket.ClientSocket.

    Attributes not specified in the constructor:
    adapter       -- ibclientpy.client_adapter.ClientAdapter object that
                     provides access to the ibapipy ClientSocket
    next_id       -- next available request ID
    id_contracts  -- dictionary of contracts by request ID

    """

    def __init__(self, loop=None):
        """Initialize a new instance of a Client.

        Keyword arguments:
        loop -- asyncio event loop

        """
        self.loop = loop
        self.adapter = ibca.ClientAdapter(self, loop)
        self.next_id = -1
        self.id_contracts = {}
        self.order_handler = iboh.OrderHandler(self)
        self.history_pending = []

    # *************************************************************************
    # Connection
    # *************************************************************************

    @asyncio.coroutine
    def connect(self, host=config.HOST, port=config.PORT,
                client_id=config.CLIENT_ID):
        """Connect to the remote TWS.

        Keyword arguments:
        host      -- host name or IP address of the TWS machine
        port      -- port number on the TWS machine
        client_id -- number used to identify this client connection

        """
        if self.is_connected():
            return
        yield from self.adapter.connect(host, port, client_id)
        # Automatically associate newly opened TWS orders with this client
        if client_id == 0:
            yield from self.adapter.req_auto_open_orders(True)
        # Wait for the next ID to get updated
        self.next_id = yield from self.get_next_valid_id()

    @asyncio.coroutine
    def disconnect(self):
        """Disconnect from the remote TWS."""
        if self.is_connected():
            yield from self.adapter.disconnect()
            self.next_id = -1

    @asyncio.coroutine
    def get_next_valid_id(self):
        """Return the next valid request ID that can be used for orders."""
        future = yield from self.adapter.req_ids(1)
        yield from future
        return future.result()

    def is_connected(self):
        """Return True if the Client is connected; False, otherwise."""
        return self.adapter.is_connected

    # *************************************************************************
    # Accounts
    # *************************************************************************

    @asyncio.coroutine
    def get_account_name(self):
        """Return the account name associated with this session as a string.

        """
        future = yield from self.adapter.req_managed_accts()
        yield from future
        return future.result()

    @asyncio.coroutine
    def get_account(self):
        """Return the ibapipy.data.account.Account instance associated with
        this session.

        """
        account_name = yield from self.get_account_name()
        accounts_list = [x.strip() for x in account_name.split(',')]
        if len(accounts_list) > 1:
            futureresults = []
            for account in accounts_list:
                if account is not "" and account.startswith('f'):
                    future = yield from self.adapter.req_account_updates(account)
                    yield from future
                    futureresults.append(future.result())
            return futureresults
        else:
            future = yield from self.adapter.req_account_updates(account_name)
            yield from future
            return future.result()

    # *************************************************************************
    # Contracts
    # *************************************************************************

    @asyncio.coroutine
    def get_contract(self, contract):
        """Return a fully populated ibapipy.data.contract.Contract instance
        from the specified Contract. The specified contract should have the
        'sec_type', 'symbol', 'currency', and 'exchange' attributes populated.

        Keyword arguments:
        contract -- ibapipy.data.contract.Contract instance

        """
        req_id = self.next_id
        self.next_id += 1
        basic_contract = get_basic_contract(contract)
        future = yield from self.adapter.req_contract_details(req_id,
                                                              basic_contract)
        yield from future
        return future.result()

    # *************************************************************************
    # Errors
    # *************************************************************************

    def on_error(self, req_id, code, message):
        """Called by the ClientAdapter when errors are reported. This
        encapsulates the following methods from the TWS API:
            - error

        Keyword arguments:
        req_id  -- request ID
        code    -- error code
        message -- error message

        """
        LOG.info('Client error/message: {0}; {1}; {2}'.format(req_id, code,
                                                              message))

    # *************************************************************************
    # Holdings
    # *************************************************************************

    def on_holding(self, contract, holding):
        """Called by the ClientAdapter when holding data is updated. This
        encapsulates the following methods from the TWS API:
            - update_portfolio

        Keyword arguments:
        contract -- ibapipy.data.contract.Contract object
        holding  -- ibapipy.data.holding.Holding object

        """
        pass

    # *************************************************************************
    # Orders
    # *************************************************************************

    @asyncio.coroutine
    def get_orders(self):
        """Return a tuple of ibapipy.data.order.Order instances with additional
        attributes on each order. Each order has a 'contract' field and
        'executions' field added to it where the contract is the underlying
        contract being traded and 'executions' is a list of executions
        associated with the order.

        """
        self.adapter.executions_fut = Future()
        self.adapter.orders_fut = Future()
        req_id = self.next_id
        self.next_id += 1
        yield from self.adapter.req_executions(req_id, ibef.ExecutionFilter())
        yield from self.adapter.req_all_open_orders()
        yield from self.adapter.executions_fut
        yield from self.adapter.orders_fut
        return self.adapter.orders_fut.result()

    @asyncio.coroutine
    def place_order(self, contract, order, profit_offset=0, loss_offset=0):
        """Place an order for the specified contract. If profit offset or loss
        offset is non-zero, a corresponding order will be placed after the
        parent order has been filled.

        The sign of the profit/loss offsets does not matter. Profit targets
        will always be placed above the entry price for long positions and
        below the entry price for short positions. Loss targets will always be
        placed below the entry price for long positions and above the entry
        price for short positions.

        Keyword arguments:
        contract      -- ibapipy.data.contract.Contract object
        parent        -- ibapipy.data.order.Order object
        profit_offset -- profit target offset from parent's fill price
                         (default: 0)
        loss_offset   -- loss target offset from parent's fill price
                         (default: 0)

        """
        req_id = self.next_id
        self.next_id += 1
        self.id_contracts[req_id] = contract
        order.order_id = req_id
        self.order_handler.add_order(order, profit_offset, loss_offset)
        yield from self.adapter.place_order(req_id, contract, order)
        return req_id

    @asyncio.coroutine
    def cancel_order(self, req_id):
        """Cancel the order associated with the specified request ID.

        Keyword arguments:
        req_id -- request ID

        """
        self.adapter.order_cancel_fut = Future()
        yield from self.adapter.cancel_order(req_id)
        yield from self.adapter.order_cancel_fut

    # *************************************************************************
    # Pricing
    # *************************************************************************

    @asyncio.coroutine
    def get_next_history_block(self, contract, start, end, timezone):
        """Return the next available block of historical ticks for the
        specified contract and time period. The result will be of the form
        (int, tuple) where "int" is the number of blocks remaining to be filled
        and "tuple" is a list of historical ticks in the current block of
        prices.

        If no more historical blocks are available, (0, None) will be
        returned.

        There will be intermittent delays in the generated data as needed to
        prevent IB pacing violations.

        Keyword arguments:
        contract   -- ibapipy.data.contract.Contract object
        start_date -- start date in 'yyyy-mm-dd hh:mm' format
        end_date   -- end date in 'yyyy-mm-dd hh:mm' format
        timezone   -- timezone in 'Country/Region' format

        """
        # Grab from the queue if we have data waiting
        if self.adapter.history_queue.qsize() > 0:
            ticks = yield from self.adapter.history_queue.get()
            return len(self.history_pending), ticks
        # Build the initial request if needed
        if len(self.history_pending) == 0:
            blocks = ibhd.get_parameters(contract, start, end, timezone)
            self.history_pending.extend(blocks)
        # Retrieve the history for the next set of parameters
        parms = self.history_pending.pop(0)
        yield from asyncio.sleep(parms['delay'])
        req_id = self.next_id
        self.next_id += 1
        self.id_contracts[req_id] = parms['contract']
        yield from self.adapter.req_historical_data(
            req_id, get_basic_contract(parms['contract']),
            parms['end_date_time'], parms['duration_str'],
            parms['bar_size_setting'], parms['what_to_show'],
            parms['use_rth'], parms['format_date'])
        ticks = yield from self.adapter.history_queue.get()
        # Check if we're done with the overall request
        if len(self.history_pending) == 0:
            yield from self.adapter.history_queue.put(None)
        return len(self.history_pending), ticks

    @asyncio.coroutine
    def cancel_history(self):
        """Stop receiving ticks from the get_next_history_block() method."""
        self.history_pending.clear()
        while self.adapter.history_queue.qsize() > 0:
            yield from self.adapter.history_queue.get()
        yield from self.adapter.history_queue.put(None)

    @asyncio.coroutine
    def get_next_tick(self, contract):
        """Return the next available realtime tick for the specified contract.
        If no more ticks are available (e.g. cancel_ticks() has been called),
        None will be returned.

        Keyword arguments:
        contract -- ibapipy.data.contract.Contract instance

        """
        key = '{0}.{1}'.format(contract.symbol, contract.currency)
        # Get the request ID for the active market data stream
        if key in self.adapter.market_data_ids:
            req_id = self.adapter.market_data_ids[key]
        # Create a new market data request
        else:
            req_id = self.next_id
            self.next_id += 1
            self.id_contracts[req_id] = contract
            yield from self.adapter.req_mkt_data(req_id, contract)
        # Pull from the queue
        tick = yield from self.adapter.tick_queue[req_id].get()
        # If the tick is None, we're done so remove the old request ID
        if tick is None:
            self.adapter.market_data_ids.pop(key)
        return tick

    @asyncio.coroutine
    def cancel_ticks(self, contract):
        """Stop receiving ticks from the get_next_tick() method."""
        key = '{0}.{1}'.format(contract.symbol, contract.currency)
        if key in self.adapter.market_data_ids:
            req_id = self.adapter.market_data_ids[key]
            yield from self.adapter.cancel_mkt_data(req_id)
            yield from self.adapter.tick_queue[req_id].put(None)

def get_basic_contract(contract):
    """TWS is complaining if we provide a fully populated contract. Here we
    we strip away everything but the core attributes.

    Keyword arguments:
    contract -- ibapipy.data.contract.Contract object

    """
    result = ibc.Contract()
    result.sec_type = contract.sec_type
    result.symbol = contract.symbol
    result.currency = contract.currency
    result.exchange = contract.exchange
    return result


text:
---
layout: post
title: kafka source code2
categories: [kafka]
keywords: kafka
---

## Kafka 0.8.2 新的offset管理

之前Kafka存在的一个非常大的性能问题(隐患)就是利用ZK来记录各个Consumer Group的消费进度。当然JVM Client帮我们自动做了这些事情,
但是Consumer需要和ZK频繁交互,而利用ZK Client API对ZK频繁写入是一个低效的操作,并且从水平扩展性上来讲也存在问题。所以ZK抖一抖,
集群吞吐量就跟着一起抖,严重的时候简直抖的停不下来。 当然某些非JVM端的API压根就不把offset存在ZK中,如基于Go的sarama,
直接自己维护了一个变量在内存中记录,总之就是问题多多了。所以非JVM的客户端,大家悠着点用。

0.8.2 Kafka引入了一个叫native offset storage的玩意儿,将offset管理从ZK移出,并且可以做到水平扩展。看到这个消息大家基本都泪流满面了。

实现原理其实也很自然,利用了Kafka自己的compacted topic,以consumer group,topic与Partition的组合作为key。
所以offset的提交就直接写到上述说的compacted topic中了,但是又由于offset是如此的重要以至于绝逼不能丢数据,
所以消息的acking级别(由request.required.acks控制)设置为-1,producer等到所有ISR收到消息后才会得到ack(数据安全性最好,但是速度会有影响)。
所以Kafka又在内存中维护了<consumer group,topic,partition>的三元组来维护最新的offset信息,consumer来取最新offset信息的时候直接内存里拿即可。
当然,敏锐性强的朋友一定想到了,kafka允许你快速的checkpoint最新的offset信息到磁盘上。

## Kafka 项目结构

**admin**      --管理员模块,操作和管理topic,paritions相关,包含create,delete topic,扩展	             		patitions

**Api**       --该模块主要负责交互数据的组装,客户端与服务端交互数据编解码

**client**    --该模块比较简单,就一个类,Producer读取kafka broker元数据信息,topic和partitions,以及leader

**cluster**   --该模块包含几个实体类,Broker,Cluster,Partition,Replica,解释他们之间关系:

Cluster由多个broker组成,一个Broker包含多个partition,一个topic的所有partitions分布在不同broker的中,一个Replica包含多个Partition。

**common**     --通用模块,只包含异常类和错误验证

**consumer**    --consumer处理模块,负责所有客户端消费者数据和逻辑处理

**controller**  --负责中央控制器选举,partition的leader选举,副本分配,副本重新分配,partition和replica扩容。

**javaapi**      --提供java的producer和consumer接口api

**log**          --Kafka文件存储模块,负责读写所有kafka的topic消息数据。

**message**    --封装多个消息组成一个"消息集"或压缩消息集。

**metrics**    --内部状态的监控模块

**network**           --网络事件处理模块,负责处理和接收客户端连接

**producer**    --producer实现模块,包括同步和异步发送消息。

**serializer**  --序列化或反序列化当前消息

**kafka**           --kafka门面入口类,副本管理,topic配置管理,leader选举实现(由contoroller模块调用)。

**tools**           --一看这就是工具模块,包含内容比较多:

1. a. 导出对应consumer的offset值.
2. b. 导出LogSegments信息,当前topic的log写的位置信息.
3. c.导出zk上所有consumer的offset值.
4. d.修改注册在zk的consumer的offset值.
5. f.producer和consumer的使用例子.

**utils**		  --Json工具类,Zkutils工具类,Utils创建线程工具类,KafkaScheduler公共调度器类,公共日志类等等。

kafka中KafkaServer类,采用门面模式,是网络处理,io处理等得入口.

ReplicaManager    副本管理

KafkaApis    处理所有request的Proxy类,根据requestKey决定调⽤用具体的handler

KafkaRequestHandlerPool 处理request的线程池,请求处理池  <-- num.io.threads io线程数量

LogManager    kafka文件存储系统管理,负责处理和存储所有Kafka的topic的partiton数据

TopicConfigManager  监听此zk节点的⼦子节点/config/changes/,通过LogManager更新topic的配置信息,topic粒度配置管理,具体请查看topic级别配置

KafkaHealthcheck 监听zk session expire,在zk上创建broker信息,便于其他broker和consumer获取其信息

KafkaController  kafka集群中央控制器选举,leader选举,副本分配。

KafkaScheduler  负责副本管理和日志管理调度等等

ZkClient         负责注册zk相关信息.

BrokerTopicStats  topic信息统计和监控

ControllerStats          中央控制器统计和监控

### Kafka server startup process

```java
def startup() {
    info("starting")
    isShuttingDown = new AtomicBoolean(false)
    shutdownLatch = new CountDownLatch(1)

    /* start scheduler */
    kafkaScheduler.startup()

    /* setup zookeeper */
    zkClient = initZk()

    /* start log manager */
    logManager = createLogManager(zkClient)
    logManager.startup()

    socketServer = new SocketServer(config.brokerId,
                                    config.hostName,
                                    config.port,
                                    config.numNetworkThreads,
                                    config.queuedMaxRequests,
                                    config.socketSendBufferBytes,
                                    config.socketReceiveBufferBytes,
                                    config.socketRequestMaxBytes)
    socketServer.startup()

    replicaManager = new ReplicaManager(config, time, zkClient, kafkaScheduler, logManager, isShuttingDown)
    kafkaController = new KafkaController(config, zkClient)

    /* start processing requests */
    apis = new KafkaApis(socketServer.requestChannel, replicaManager, zkClient, config.brokerId, config, kafkaController)
    requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads)

    Mx4jLoader.maybeLoad()

    replicaManager.startup()

    kafkaController.startup()

    topicConfigManager = new TopicConfigManager(zkClient, logManager)
    topicConfigManager.startup()

    /* tell everyone we are alive */
    kafkaHealthcheck = new KafkaHealthcheck(config.brokerId, config.advertisedHostName, config.advertisedPort, config.zkSessionTimeoutMs, zkClient)
    kafkaHealthcheck.startup()

    registerStats()
    startupComplete.set(true);
    info("started")
  }

  private def initZk(): ZkClient = {
    info("Connecting to zookeeper on " + config.zkConnect)
    val zkClient = new ZkClient(config.zkConnect, config.zkSessionTimeoutMs, config.zkConnectionTimeoutMs, ZKStringSerializer)
    ZkUtils.setupCommonPaths(zkClient)
    zkClient
  }

  /**
   *  Forces some dynamic jmx beans to be registered on server startup.
   */
  private def registerStats() {
    BrokerTopicStats.getBrokerAllTopicsStats()
    ControllerStats.uncleanLeaderElectionRate
    ControllerStats.leaderElectionTimer
  }
```

## Consumer 逻辑

### Consumer group 的注意事项

Consumer Group是整个Kafka集群全局的. 所以要特别小心在新的逻辑启动之前要关闭所有的旧的逻辑(消费者进程).

当新的消费者加入同一个消费组时,Kafka会添加这个消费者的线程到要消费的topic的可用线程集合中,并且触发re-balance.

在re-balance时,kafka会分配可用的partition给可用的线程,可能移动一个partition到其他的线程中.

如果你的消费者逻辑混合了新的和旧的处理逻辑,很可能有些消息会被分配到旧的处理逻辑中.

High Level Consumer可以(应该)是多线程的应用程序.线程模型是以topic的partitions数量为中心的,不过有些规则:

1. 如果线程数量多于partition的数量,有部分线程无法消费该topic下任何一条消息
2. 如果线程数量少于partition的数量,有一些线程会消费多个partition的数据
3. 如果线程数量等于partition的数量,则正好一个线程消费一个partition的数据
4. 当添加更多的消费者进程/线程会触发re-balance,导致partition的分配发生了变化

### 使用

单线程版本

```scala
  def createConsumers(zookeeperConnect: String, consumerGroup: String, sessionTimeout: String,
      syncTime: String, commitInterval: String, connectionTimeout: String, consumerTimeout: String,
      autoOffsetReset: String, topic: String, numOfConsumers: Int): Array[KafkaStream[Array[Byte], Array[Byte]]] = {

    val props: Properties = new Properties
    props.put("zookeeper.connect", zookeeperConnect)
    props.put("group.id", consumerGroup)
    props.put("zookeeper.session.timeout.ms", sessionTimeout)
    props.put("zookeeper.sync.time.ms", syncTime)
    props.put("auto.commit.interval.ms", commitInterval)
    props.put("zookeeper.connection.timeout.ms", connectionTimeout)
    props.put("consumer.timeout.ms", consumerTimeout)
    props.put("auto.offset.reset", autoOffsetReset)
    val config = new ConsumerConfig(props)
    val consumerConnector = kafka.consumer.Consumer.createJavaConsumerConnector(config)
    val topicCountMap: util.Map[String, Integer] = new util.HashMap[String, Integer]()
    topicCountMap.put(topic, new Integer(numOfConsumers))
    val consumerMap = consumerConnector.createMessageStreams(topicCountMap)
    consumerMap.get(topic).asScala.toArray
  }

  val streamIterator = consumer map (_.iterator())
  streamIterator foreach { ci: ConsumerIterator[Array[Byte], Array[Byte]] => {

    log.info("stream query foreach")

    if (System.currentTimeMillis() < requestEndTime && KafkaUtils.timedHasNext(ci)) {

        //if next throw exception, actor returns immediately
        val message = ci.next()

        val key = if(message.key == null) null else new String(message.key());
        val value = new String(message.message())

        log.debug(s"consumer actor get message: ($key, $value) for topic: $topic")
        lstbf += MessageContent(key, value)    }
```

**多线程版本**

```java
public class ConsumerTest implements Runnable {
    private KafkaStream m_stream;
    private int m_threadNumber;

    public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
        m_threadNumber = a_threadNumber;
        m_stream = a_stream;
    }

    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext())
            System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));
        System.out.println("Shutting down Thread: " + m_threadNumber);
    }
}

public void run(int a_numThreads) {
    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, new Integer(a_numThreads));
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

    // now launch all the threads
    //
    executor = Executors.newFixedThreadPool(a_numThreads);

    // now create an object to consume the messages
    //
    int threadNumber = 0;
    for (final KafkaStream stream : streams) {
        executor.execute(new ConsumerTest(stream, threadNumber)); // 开始执行
        threadNumber++;
    }
}
```

### Consumer

Consumer定义在ConsumerConnector接口同一个文件中.默认创建的ConsumerConnector是ZookeeperConsumerConnector

ConsumerConnector主要有创建消息流(createMessageStreams)和提交offset(commitOffsets)两种方法

Consumer会根据消息流消费数据, 并且定时提交offset.由客户端自己保存offset是kafka采用pull拉取消息的一个附带工作.

```scala
trait ConsumerConnector {
  def createMessageStreams(topicCountMap: Map[String,Int]): Map[String, List[KafkaStream[Array[Byte],Array[Byte]]]]
  def commitOffsets(offsetsToCommit: immutable.Map[TopicAndPartition, OffsetAndMetadata], retryOnFailure: Boolean)
  def setConsumerRebalanceListener(listener: ConsumerRebalanceListener)
  def shutdown()
}
```

一个Consumer会创建一个ZookeeperConsumerConnector,代表一个消费者进程.

```scala
private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig, val enableFetcher: Boolean)
        extends ConsumerConnector with Logging with KafkaMetricsGroup {
  private var fetcher: Option[ConsumerFetcherManager] = None
  private var zkUtils: ZkUtils = null
  private var topicRegistry = new Pool[String, Pool[Int, PartitionTopicInfo]]
  private val checkpointedZkOffsets = new Pool[TopicAndPartition, Long]
  private val topicThreadIdAndQueues = new Pool[(String, ConsumerThreadId), BlockingQueue[FetchedDataChunk]]
  private val scheduler = new KafkaScheduler(threads = 1, threadNamePrefix = "kafka-consumer-scheduler-")
  private val messageStreamCreated = new AtomicBoolean(false)
  private var offsetsChannel: BlockingChannel = null
  private var sessionExpirationListener: ZKSessionExpireListener = null
  private var topicPartitionChangeListener: ZKTopicPartitionChangeListener = null
  private var loadBalancerListener: ZKRebalancerListener = null
  private var wildcardTopicWatcher: ZookeeperTopicEventWatcher = null
  private var consumerRebalanceListener: ConsumerRebalanceListener = null

  connectZk()                       // ① 创建ZkUtils,会创建对应的ZkConnection和ZkClient
  createFetcher()                   // ② 创建ConsumerFetcherManager,消费者fetcher线程
  ensureOffsetManagerConnected()    // ③ 确保连接上OffsetManager.
  if (config.autoCommitEnable) {    // ④ 启动定时提交offset线程
    scheduler.startup
    scheduler.schedule("kafka-consumer-autocommit", autoCommit, ...)
  }
}
```

**fetcher:** 消费者获取数据, 使用ConsumerFetcherManager fetcher线程抓取数据

**zkUtils:** 消费者要和ZK通信, 除了注册自己,还有其他信息也会写到ZK中

**topicThreadIdAndQueues:** 消费者会指定自己消费哪些topic,并指定线程数, 所以topicThreadId都对应一个队列

**messageStreamCreated:** 消费者会创建消息流, 每个队列都对应一个消息流

**offsetsChannel:** offset可以存储在ZK或者kafka中,如果存在kafka里,像其他请求一样,需要和Broker通信

还有其他几个Listener监听器,分别用于 topicPartition 的更新,负载均衡,消费者重新负载等

虽然high level consumer不需要在应用程序中自己管理offset等,但kafka内部还是会帮你管理offset的.

为什么ConsumerConnector是ZookeeperConsumerConnector,因为消费者的offset是保存在ZooKeeper中的.

broker 在 zookeeper 中的信息

```
[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 1] ls /brokers
[topics, ids]

[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 2] ls /brokers/ids
[3, 5, 4]
[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 4] get /brokers/ids/3
{"jmx_port":10055,"timestamp":"1453380999577","host":"192.168.48.153","version":1,"port":9092}
```

```
[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 17] get /brokers/topics/topic1
{"version":1,"partitions":{"2":[5,4],"1":[4,3],"0":[3,5]}}   ⬅

[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 12] ls /brokers/topics/topic1/partitions
[2, 1, 0]
[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 13] ls /brokers/topics/topic1/partitions/0
[state]
[zk: 192.168.47.83:2181,192.168.47.84:2181,192.168.47.86:2181(CONNECTED) 15] get /brokers/topics/topic1/partitions/0/state
{"controller_epoch":1775,"leader":3,"version":1,"leader_epoch":145,"isr":[3,5]}
```

partition0 的 leader 是 3, 复制在 broker 3 和 5 中

**Consumer id registry:** /consumers/[group_id]/ids/[consumer_id] -> topic1,...topicN

每个消费者会将它的id注册为临时znode并且将它所消费的topic设置为znode的值,当客户端(消费者)退出时,znode(consumer_id)会被删除

**Partition owner registry:** /consumers/[group_id]/owner/[topic]/[broker_id-partition_id] --> consumer_node_id

在消费时,每个topic的partition只能被一个消费者组中的唯一的一个消费者消费.在每次重新负载的时候,这个映射策略就会重新构建

**Consumer offset tracking:** /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] --> offset_counter_value

每个消费者都要跟踪自己消费的每个Partition最近的offset.表示自己读取到Partition的最新位置.

由于一个Partition只能被消费组中的一个消费者消费,所以offset是以消费组为级别的,而不是消费者.

因为如果原来的消费者挂了后,应当将这个Partition交给同一个消费组中别的消费者,而此时offset是没有变化的.

一个partition可以被不同的消费者组中的不同消费者消费,所以不同的消费者组必须维护他们各自对该partition消费的最新的offset

### 初始化

1. 因为消费者要和ZK通信,所以connectZk会确保连接上ZooKeeper
2. 消费者要消费数据,需要有抓取线程,所有的抓取线程交给ConsumerFetcherManager统一管理
3. 由消费者客户端自己保存offset,而消费者会消费多个topic的多个partition.
4. 类似多个数据抓取线程有管理类,多个partition的offset管理类OffsetManager是一个GroupCoordinator
5. 定时提交线程会使用OffsetManager建立的通道定时提交offset到zk或者kafka.

### 消费

consume并不真正的消费数据,只是初始化存放数据的 queue.真正消费数据的是对该queue进行shallow iterator.

在kafka的运行过程中,会有其他的线程将数据放入partition对应的queue中. 而queue是用于KafkaStream的.

一旦数据添加到queue后,KafkaStream的阻塞队列就有数据了,消费者就可以从队列中消费消息.

```java
def createMessageStreams[K,V](topicCountMap: Map[String,Int],
  keyDecoder: Decoder[K], valueDecoder: Decoder[V]) : Map[String, List[KafkaStream[K,V]]] = {
  consume(topicCountMap, keyDecoder, valueDecoder)
}

def consume[K, V](topicCountMap: scala.collection.Map[String,Int],
  keyDecoder: Decoder[K], valueDecoder: Decoder[V]) : Map[String,List[KafkaStream[K,V]]] = {

  val topicCount = TopicCount.constructTopicCount(consumerIdString, topicCountMap)
  val topicThreadIds = topicCount.getConsumerThreadIdsPerTopic

  // make a list of (queue,stream) pairs, one pair for each threadId 只是准备了队列和流,数据什么时候填充呢?
  val queuesAndStreams = topicThreadIds.values.map(threadIdSet =>
    threadIdSet.map(_ => {
      val queue =  new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
      val stream = new KafkaStream[K,V](queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId)
      (queue, stream)
    })
  ).flatten.toList  //threadIdSet是个集合,外层的topicThreadIds.values也是集合,所以用flatten压扁为queue-stream对

  val dirs = new ZKGroupDirs(config.groupId)                  // /consumers/[group_id]
  registerConsumerInZK(dirs, consumerIdString, topicCount)    // /consumers/[group_id]/ids/[consumer_id]
  reinitializeConsumer(topicCount, queuesAndStreams)          // 重新初始化消费者 ⬅

  // 返回KafkaStream, 每个Topic都对应了多个KafkaStream. 数量和topicCount中的count一样
  loadBalancerListener.kafkaMessageAndMetadataStreams.asInstanceOf[Map[String, List[KafkaStream[K,V]]]]
}
```

consumerIdString会返回当前Consumer在哪个ConsumerGroup的编号.每个consumer在消费组中的编号都是唯一的.

一个消费者,对一个topic可以使用多个线程一起消费(一个进程可以有多个线程). 当然一个消费者也可以消费多个topic.

```java
def makeConsumerThreadIdsPerTopic(consumerIdString: String, topicCountMap: Map[String,  Int]) = {
  val consumerThreadIdsPerTopicMap = new mutable.HashMap[String, Set[ConsumerThreadId]]()

  for ((topic, nConsumers) <- topicCountMap) {                // 每个topic有几个消费者线程
    val consumerSet = new mutable.HashSet[ConsumerThreadId]   // 一个消费者线程对应一个ConsumerThreadId

    for (i <- 0 until nConsumers)
      consumerSet += ConsumerThreadId(consumerIdString, i)
      consumerThreadIdsPerTopicMap.put(topic, consumerSet)      // 每个topic都有多个Consumer线程,但是只有一个消费者进程
  }

  consumerThreadIdsPerTopicMap                                // topic到消费者线程集合的映射
}
```

对于消费者而言,它只要指定要消费的topic和线程数量就可以了,其他具体这个topic分成多少个partition,
以及topic-partition是分布是哪个broker上,对于客户端而言都是透明的.
客户端关注的是我的每个线程都对应了一个队列,每个队列都是一个消息流就可以了.

在Producer以及前面分析的Fetcher,都是以Broker-Topic-Partition为级别的.
AbstractFetcherManager的fetcherThreadMap就是以brokerAndFetcherId来创建拉取线程的.
而消费者是通过拉取线程才有数据可以消费的,所以客户端的每个线程实际上也是针对Partition级别的.

### registerConsumerInZK

消费者需要向ZK注册一个临时节点,路径为:/consumers/[group_id]/ids/[consumer_id],内容为订阅的topic.

```java
private def registerConsumerInZK(dirs: ZKGroupDirs, consumerIdString: String, topicCount: TopicCount) {
  val consumerRegistrationInfo = Json.encode(Map("version" -> 1,
    "subscription" -> topicCount.getTopicCountMap, "pattern" -> topicCount.pattern, "timestamp" -> timestamp))
  val zkWatchedEphemeral = new ZKCheckedEphemeral(dirs.consumerRegistryDir + "/" + consumerIdString,
    consumerRegistrationInfo, zkUtils.zkConnection.getZookeeper, false)
  zkWatchedEphemeral.create()
}
```

当前Consumer在ZK注册之后,需要重新初始化Consumer. 对于全新的消费者,注册多个监听器,在zk的对应节点的注册事件发生时,会回调监听器的方法.

当前Consumer在ZK注册之后,需要重新初始化Consumer. 对于全新的消费者,注册多个监听器,在zk的对应节点的注册事件发生时,会回调监听器的方法.
将topic对应的消费者线程id及对应的LinkedBlockingQueue放入topicThreadIdAndQueues中,LinkedBlockingQueue是真正存放数据的queue

1. 注册sessionExpirationListener,监听状态变化事件.在session失效重新创建session时调用
2. 向/consumers/[group_id]/ids注册Child变更事件的loadBalancerListener,当消费组下的消费者发生变化时调用
3. 向/brokers/topics/[topic]注册Data变更事件的topicPartitionChangeListener,在topic数据发生变化时调用

显式调用loadBalancerListener.syncedRebalance(), 会调用reblance方法进行consumer的初始化工作

```java
private def reinitializeConsumer[K,V](topicCount: TopicCount,
  queuesAndStreams: List[(LinkedBlockingQueue[FetchedDataChunk],KafkaStream[K,V])]) {
  val dirs = new ZKGroupDirs(config.groupId)
  // ② listener to consumer and partition changes
  if (loadBalancerListener == null) {
    val topicStreamsMap = new mutable.HashMap[String,List[KafkaStream[K,V]]]
    loadBalancerListener = new ZKRebalancerListener(config.groupId, consumerIdString,
      topicStreamsMap.asInstanceOf[scala.collection.mutable.Map[String, List[KafkaStream[_,_]]]])
  }
  // ① create listener for session expired event if not exist yet
  if (sessionExpirationListener == null) sessionExpirationListener =
    new ZKSessionExpireListener(dirs, consumerIdString, topicCount, loadBalancerListener)
  // ③ create listener for topic partition change event if not exist yet
  if (topicPartitionChangeListener == null)
    topicPartitionChangeListener = new ZKTopicPartitionChangeListener(loadBalancerListener)

  // listener to consumer and partition changes
  zkUtils.zkClient.subscribeStateChanges(sessionExpirationListener)
  zkUtils.zkClient.subscribeChildChanges(dirs.consumerRegistryDir, loadBalancerListener)
  // register on broker partition path changes.
  topicStreamsMap.foreach { topicAndStreams =>
    zkUtils.zkClient.subscribeDataChanges(BrokerTopicsPath+"/"+topicAndStreams._1, topicPartitionChangeListener)
  }

  // explicitly trigger load balancing for this consumer
  loadBalancerListener.syncedRebalance()
}
```

ZKRebalancerListener传入ZKSessionExpireListener和ZKTopicPartitionChangeListener.它们都会使用ZKRebalancerListener完成自己的工作.

因为消费者加入/退出时,消费组的成员会发生变化,而消费组中的所有存活消费者负责消费可用的partitions.
可用的partitions或者消费组中的消费者成员一旦发生变化,都要重新分配partition给存活的消费者.下面是一个示例.

当然分配partition的工作绝不仅仅是这么简单的,还要处理与之相关的线程,并重建必要的数据:

1. 关闭数据抓取线程,获取之前为topic设置的存放数据的queue并清空该queue
2. 释放 partition的ownership, 删除 partition和 consumer 的对应关系
3. 为各个 partition 重新分配 threadId
   获取 partition 最新的 offset 并重新初始化新的 PartitionTopicInfo(queue存放数据,两个offset为partition最新的offset)
4. 重新将 partition 对应的新的 consumer 信息写入 zookeeper
5. 重新创建 partition 的 fetcher 线程

```scala
private def rebalance(cluster: Cluster): Boolean = {
  val myTopicThreadIdsMap = TopicCount.constructTopicCount(group, consumerIdString,
    zkUtils, config.excludeInternalTopics).getConsumerThreadIdsPerTopic

  val brokers = zkUtils.getAllBrokersInCluster()

  if (brokers.size == 0) {
    zkUtils.zkClient.subscribeChildChanges(BrokerIdsPath, loadBalancerListener)
    true
  } else {
    // ① 停止fetcher线程防止数据重复.如果当前调整失败了,被释放的partitions可能被其他消费者拥有.
    // 而没有先停止fetcher的话,原先的消费者仍然会和新的拥有者共同消费同一份数据.
    closeFetchers(cluster, kafkaMessageAndMetadataStreams, myTopicThreadIdsMap)
    // ② 释放topicRegistry中topic-partition的owner
    releasePartitionOwnership(topicRegistry)
    // ③ 为partition重新分配消费者....
    // ④ 为partition添加consumer owner
    if(reflectPartitionOwnershipDecision(partitionAssignment)) {
        allTopicsOwnedPartitionsCount = partitionAssignment.size
        topicRegistry = currentTopicRegistry
        // ⑤ 创建拉取线程
        updateFetcher(cluster)
        true
    }
  }
}
```

rebalance操作涉及了以下内容:

1. PartitionOwnership: Partition的所有者(ownership)的删除和重建
2. AssignmentContext: 分配信息上下文
3. PartitionAssignor: 为Partition分配Consumer的算法
4. PartitionAssignment: Partition分配之后的上下文
5. PartitionTopicInfo: Partition的最终信息
6. Fetcher: 完成了rebalance,消费者就可以重新开始抓取数据

### PartitionOwnership

topicRegistry的数据结构是: topic -> (partition -> PartitionTopicInfo), 表示现有的topic注册信息.

当partition被consumer所拥有后, 会在zk中创建/consumers/[group_id]/owner/[topic]/[partition_id] --> consumer_node_id
释放所有partition的ownership, 数据来源于topicRegistry的topic-partition(消费者所属的group_id也是确定的).

所以deletePartitionOwnershipFromZK会删除/consumers/[group_id]/owner/[topic]/[partition_id]节点.

这样partition没有了owner,说明这个partition不会被consumer消费了,也就相当于consumer释放了partition.

```scala
private def releasePartitionOwnership(localTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]])= {
  for ((topic, infos) <- localTopicRegistry) {
    for(partition <- infos.keys) {
      deletePartitionOwnershipFromZK(topic, partition)
    }
    localTopicRegistry.remove(topic)
  }
  allTopicsOwnedPartitionsCount = 0
}

private def deletePartitionOwnershipFromZK(topic: String, partition: Int) {
  val topicDirs = new ZKGroupTopicDirs(group, topic)
  val znode = topicDirs.consumerOwnerDir + "/" + partition
  zkUtils.deletePath(znode)
}
```

重建ownership. 参数partitionAssignment会指定partition(TopicAndPartition)要分配给哪个consumer(ConsumerThreadId)消费的.

```scala
private def reflectPartitionOwnershipDecision(partitionAssignment: Map[TopicAndPartition, ConsumerThreadId]): Boolean = {
  var successfullyOwnedPartitions : List[(String, Int)] = Nil
  val partitionOwnershipSuccessful = partitionAssignment.map { partitionOwner =>
    val topic = partitionOwner._1.topic
    val partition = partitionOwner._1.partition
    val consumerThreadId = partitionOwner._2

    // 返回/consumers/[group_id]/owner/[topic]/[partition_id]节点路径,然后创建节点,节点内容为:consumerThreadId
    val partitionOwnerPath = zkUtils.getConsumerPartitionOwnerPath(group, topic, partition)
    zkUtils.createEphemeralPathExpectConflict(partitionOwnerPath, consumerThreadId.toString)

    // 成功创建的节点,加入到列表中
    successfullyOwnedPartitions ::= (topic, partition)
    true
  }
  // 判断上面的创建节点操作(为consumer分配partition)是否有错误,一旦有一个有问题,就全部回滚(删除掉).只有所有成功才算成功
  val hasPartitionOwnershipFailed = partitionOwnershipSuccessful.foldLeft(0)((sum, decision) => sum + (if(decision) 0 else 1))
  if(hasPartitionOwnershipFailed > 0) {
    successfullyOwnedPartitions.foreach(topicAndPartition => deletePartitionOwnershipFromZK(topicAndPartition._1, topicAndPartition._2))
    false
  } else true
}
```

将可用的partitions以及消费者线程排序, 将partitions处于线程数,表示每个线程(不是消费者数量)平均可以分到几个partition.
如果除不尽,剩余的会分给前面几个消费者线程. 比如有两个消费者,每个都是两个线程,一共有5个可用的partitions:(p0-p4).
每个消费者线程(一共四个线程)可以获取到至少一共partition(5/4=1),剩余一个(5%4=1)partition分给第一个线程.

最后的分配结果为: p0 -> C1-0, p1 -> C1-0, p2 -> C1-1, p3 -> C2-0, p4 -> C2-1

![](http://img.blog.csdn.net/20160127083105757)

```scala
class RangeAssignor() extends PartitionAssignor with Logging {
  def assign(ctx: AssignmentContext) = {
    val valueFactory = (topic: String) => new mutable.HashMap[TopicAndPartition, ConsumerThreadId]
    // consumerThreadId -> (TopicAndPartition -> ConsumerThreadId). 所以上面的valueFactory的参数不对哦!
    val partitionAssignment = new Pool[String, mutable.Map[TopicAndPartition, ConsumerThreadId]](Some(valueFactory))
    for (topic <- ctx.myTopicThreadIds.keySet) {
      val curConsumers = ctx.consumersForTopic(topic)                   // 订阅了topic的消费者列表(4)
      val curPartitions: Seq[Int] = ctx.partitionsForTopic(topic)       // 属于topic的partitions(5)
      val nPartsPerConsumer = curPartitions.size / curConsumers.size    // 每个线程都可以有这些个partition(1)
      val nConsumersWithExtraPart = curPartitions.size % curConsumers.size  // 剩余的分给前面几个(1)
      for (consumerThreadId <- curConsumers) {                          // 每个消费者线程: C1_0,C1_1,C2_0,C2_1
        val myConsumerPosition = curConsumers.indexOf(consumerThreadId) // 线程在列表中的索引: 0,1,2,3
        val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart)
        val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1)
        // Range-partition the sorted partitions to consumers for better locality.
        // The first few consumers pick up an extra partition, if any.
        if (nParts > 0)
          for (i <- startPart until startPart + nParts) {
            val partition = curPartitions(i)
            // record the partition ownership decision 记录partition的ownership决定,即把partition分配给consumerThreadId
            val assignmentForConsumer = partitionAssignment.getAndMaybePut(consumerThreadId.consumer)
            assignmentForConsumer += (TopicAndPartition(topic, partition) -> consumerThreadId)
          }
        }
      }
    }
    // 前面的for循环中的consumers是和当前消费者有相同topic的,如果消费者的topic和当前消费者不一样,则在本次assign中不会为它分配的.
    // assign Map.empty for the consumers which are not associated with topic partitions 没有和partition关联的consumer分配空的partiton.
    ctx.consumers.foreach(consumerId => partitionAssignment.getAndMaybePut(consumerId))
    partitionAssignment
  }
}
```


text: Events page


text:
package org.apache.flink.runtime.io.network.netty;

import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.NetworkClientHandler;
import org.apache.flink.runtime.io.network.PartitionRequestClient;
import org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;

import org.apache.flink.shaded.netty4.io.netty.channel.Channel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.flink.runtime.concurrent.FutureUtils.completeFromCallable;

/**
 * Factory for {@link NettyPartitionRequestClient} instances.
 *
 * <p>Instances of partition requests clients are shared among several {@link RemoteInputChannel}
 * instances.
 */
class PartitionRequestClientFactory {
	private static final Logger LOG = LoggerFactory.getLogger(PartitionRequestClientFactory.class);

	private final NettyClient nettyClient;

	private final int retryNumber;

	private final ConcurrentMap<ConnectionID, CompletableFuture<NettyPartitionRequestClient>> clients = new ConcurrentHashMap<>();

	PartitionRequestClientFactory(NettyClient nettyClient) {
		this(nettyClient, 0);
	}

	PartitionRequestClientFactory(NettyClient nettyClient, int retryNumber) {
		this.nettyClient = nettyClient;
		this.retryNumber = retryNumber;
	}

	/**
	 * Atomically establishes a TCP connection to the given remote address and
	 * creates a {@link NettyPartitionRequestClient} instance for this connection.
	 */
	NettyPartitionRequestClient createPartitionRequestClient(ConnectionID connectionId) throws IOException, InterruptedException {
		while (true) {
			AtomicBoolean isTheFirstOne = new AtomicBoolean(false);
			CompletableFuture<NettyPartitionRequestClient> clientFuture = clients.computeIfAbsent(connectionId, unused -> {
				isTheFirstOne.set(true);
				return new CompletableFuture<>();
			});
			if (isTheFirstOne.get()) {
				completeFromCallable(clientFuture, () -> connectWithRetries(connectionId));
			}

			final NettyPartitionRequestClient client;
			try {
				client = clientFuture.get();
			} catch (ExecutionException e) {
				throw new IOException(e);
			}
			// Make sure to increment the reference count before handing a client
			// out to ensure correct bookkeeping for channel closing.
			if (client.incrementReferenceCounter()) {
				return client;
			} else {
				destroyPartitionRequestClient(connectionId, client);
			}
		}
	}

	private NettyPartitionRequestClient connectWithRetries(ConnectionID connectionId) {
		int tried = 0;
		while (true) {
			try {
				return connect(connectionId);
			} catch (RemoteTransportException e) {
				tried++;
				LOG.error("Failed {} times to connect to {}", tried, connectionId.getAddress(), e);
				if (tried > retryNumber) {
					throw new CompletionException(e);
				}
			}
		}
	}

	private NettyPartitionRequestClient connect(ConnectionID connectionId) throws RemoteTransportException {
		try {
			Channel channel = nettyClient.connect(connectionId.getAddress()).await().channel();
			NetworkClientHandler clientHandler = channel.pipeline().get(NetworkClientHandler.class);
			return new NettyPartitionRequestClient(channel, clientHandler, connectionId, this);
		} catch (Exception e) {
			throw new RemoteTransportException(
				"Connecting to remote task manager '" + connectionId.getAddress() +
					"' has failed. This might indicate that the remote task " +
					"manager has been lost.",
				connectionId.getAddress(), e);
		}
	}

	void closeOpenChannelConnections(ConnectionID connectionId) {
		CompletableFuture<NettyPartitionRequestClient> entry = clients.get(connectionId);

		if (entry != null && !entry.isDone()) {
			entry.thenAccept(client -> {
				if (client.disposeIfNotUsed()) {
					clients.remove(connectionId, entry);
				}
			});
		}
	}

	int getNumberOfActiveClients() {
		return clients.size();
	}

	/**
	 * Removes the client for the given {@link ConnectionID}.
	 */
	void destroyPartitionRequestClient(ConnectionID connectionId, PartitionRequestClient client) {
		final CompletableFuture<NettyPartitionRequestClient> future = clients.get(connectionId);
		if (future != null && future.isDone()) {
			future.thenAccept(futureClient -> {
				if (client.equals(futureClient)) {
					clients.remove(connectionId, future);
				}
			});
		}
	}

}


text:
package com.sun.xml.internal.ws.spi.db;

import java.lang.reflect.Method;
import java.security.AccessController;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;

/**
 * MethodGetter
 * @author shih-chang.chen@oracle.com
 * @exclude
 */
public class MethodGetter extends PropertyGetterBase {

    private Method method;

    public MethodGetter(Method m) {
        method = m;
        type = m.getReturnType();
    }

    public Method getMethod() {
        return method;
    }

    public <A> A getAnnotation(Class<A> annotationType) {
        Class c = annotationType;
        return (A) method.getAnnotation(c);
    }

    static class PrivilegedGetter implements PrivilegedExceptionAction {
        private Object value;
        private Method method;
        private Object instance;
        public PrivilegedGetter(Method m, Object instance) {
            super();
            this.method = m;
            this.instance = instance;
        }
        public Object run() throws IllegalAccessException {
            if (!method.isAccessible()) {
                method.setAccessible(true);
            }
            try {
                value = method.invoke(instance, new Object[0]);
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return null;
        }
    }

    public Object get(final Object instance) {
        final Object[] args = new Object[0];
        try {
            if (method.isAccessible()) {
                return method.invoke(instance, args);
            } else {
                PrivilegedGetter privilegedGetter = new PrivilegedGetter(method, instance);
                try {
                    AccessController.doPrivileged(privilegedGetter);
                } catch (PrivilegedActionException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                return privilegedGetter.value;
            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return null;
    }
}


text:
<!DOCTYPE html>

  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>HTTP/2 + EventSource JSON push notifications</title>

HTTP/2 + EventSource JSON push notifications


text:
<?xml version="1.0" encoding="utf-8"?>
<selector xmlns:android="http://schemas.android.com/apk/res/android">
    <item android:state_pressed="true"><color android:color="#778899" />
    </item>
    <item android:state_selected="true"><color android:color="#778899" />
    </item>
</selector>


text:
<!DOCTYPE html>

	<meta charset="utf-8">
	<title>XSRF example</title>

	<meta name="author" content="raesene">

	<meta name="viewport" content="width=device-width, initial-scale=1.0">

	<link href="https://raesene.github.io/assets/resources/bootstrap/css/bootstrap.min.css" rel="stylesheet">
	<link href="https://raesene.github.io/assets/resources/font-awesome/css/font-awesome.min.css" rel="stylesheet">
	<link href="https://raesene.github.io/assets/resources/syntax/syntax.css" rel="stylesheet">
	<link href="https://raesene.github.io/assets/css/style.css" rel="stylesheet">

	<link rel="alternate" type="application/rss+xml" title="" href="https://raesene.github.io/feed.xml">

			<button type="button" class="navbar-toggle" data-toggle="collapse" data-target="#bs-example-navbar-collapse-1">
				Toggle navigation

			</button>

				Raesene's Ramblings

Home

Categories

Tags

		<button type="button" class="btn btn-default dropdown-toggle" data-toggle="dropdown">

		</button>

Home

Categories

Tags

Top of Page

        Raesene's Ramblings

	Security Geek, Kubernetes, Docker, Ruby, Hillwalking

XSRF example

	   September
	   26th,

	   2006

Google Cross-site Request Forgery

Cross Site Request Forgery is one of those vulnerability classes that can be a bit tricky to explain, so it's always nice to find a decent live example.  This one's pretty harmless, just changes your google languange preferences, but I reckon that we'll see a real growth in this kind of problem over the next couple of years as "the bad guys"(tm) keep looking for innovative new ways to attack web apps.

					Web Security (39)

Share Post

          Twitter

          Facebook

          Google+

raesene

Security Geek, Kubernetes, Docker, Ruby, Hillwalking

← Previous

Next →

				© 2022 raesene with Jekyll. Theme: dbyll by dbtek.


text:
_What's happening currently_

## Current targets

### Walking

* Currently 5,000 steps per day
* 20 'heart minutes' per day
* 45 minutes of movement per day

Weight
* currently 87/88 kg
* Medium term: get back down to 84 kg - my weight before Thanksgiving
* Longer term: 80 kg

### Current devices/monitors

Google Fit app
* Google Pixel 3 XL Phone monitors walking
* Fit-Index app and ES-26M scale monitor weight, BMI etc

Standalone/unconnected
* $10 pulse oximeter - nice but no wifi

FitBit
* I like the Fitbit but lose/put in washing machine too often/has issues connecting

Current level of drive
* Very much in a 'don't break the chain' mode with my walking
* May or may not be good
* Need to add some non-walking moves

***

_What's wanted in the future_

## Building a ongoing coach/player effort

### Standing agendas for online meetups

* What are the things we should discuss?
	* Effective communication
* Written, online, collaborative
	* Tools: groups.io? github?
* Fees/Scheduling/Insurance
	* Can this work within existing frameworks?
	* Or do we create new non-sickness-plan ways of collaborating?

***

_What to plan for_

### Before exercising / Planning / Signalling

* Building up to a list of daily exercises/movement routines
	* For the next 5/10 years?
* Better ways of signaling the time for doing something
	* Too easy to ignore the bell

### While Exercising

Things list of things to think about while I am walking up and down the hallNeed
* Standing up straight, stretching neck, squeezing handles hard etc

### After exercising

* Android app so I can share Google fit data with family and friends
* Fast and easy ways of keeping data and reporting progress


text:
package cmp

type Sender interface {
	Send(interface{}) error
}

type Receiver interface {
	Receive() (interface{}, error)
}

type Communicator interface {
	Sender
	Receiver
}


text:
/*global require,console,Ti*/
/*jslint devel: true, forin: true */
/**
 * An enhanced fork of the original TiDraggable module by Pedro Enrique,
 * allows for simple creation of "draggable" views.
 *
 * Copyright (C) 2013 Seth Benjamin
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * -- Original License --
 *
 * Copyright 2012 Pedro Enrique
 *
 * 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.
 */

(function () {
    'use strict';

    var Draggable = require('ti.draggable'),
        mainWindow = Ti.UI.createWindow({
            backgroundColor : 'white',
            exitOnClose : true,
            fullscreen : true
        }),
        subscribe = function (proxy, observer) {
            var key, events, eIndex;

            for (key in observer) {
                if (typeof observer[key] === 'function') {
                    events = key.split(' ');

                    for (eIndex in events) {
                        proxy.addEventListener(events[eIndex], observer[key]);
                    }
                }
            }
        },
        createDraggableSquare = function (name, color, axis) {
            var view = Draggable.createView({
                    width : 100,
                    height : 100,
                    borderRadius : 3,
                    backgroundColor : color || 'black',
                    draggableConfig : {
                        axis : axis,
                        minLeft : 0,
                        maxLeft : Ti.Platform.displayCaps.platformWidth - 100,
                        minTop : 0,
                        maxTop : Ti.Platform.displayCaps.platformHeight - 100,
                    }
                });

            view.add(Ti.UI.createLabel({
                text : name
            }));

            subscribe(view, {
                'start move end cancel' : function (e) {
                    console.log(
                        'Event: ' + e.type,
                        'Left: ' + e.left,
                        'Top: ' + e.top
                    );
                }
            });

            return view;
        };

    mainWindow.add(createDraggableSquare('Horizontal', 'red', 'x'));
    mainWindow.add(createDraggableSquare('Vertical', 'blue', 'y'));
    mainWindow.add(createDraggableSquare('Free', 'green'));

    mainWindow.open();
}());


text:
<?php

namespace Guzzle\Openstack\Tests\Compute\Command;

use Guzzle\Openstack\Tests\Compute\Common\ComputeTestCase;
use Guzzle\Openstack\Compute\ComputeConstants;

/**
 * ListImagesDetail Tests
 */
class ListImagesDetailTest extends ComputeTestCase
{

    public function testListImagesDetail()
    {
        $this->setMockResponse($this->client, 'compute/ListImagesDetail');
        $command = $this->client->getCommand(ComputeConstants::LIST_IMAGES_DETAIL);
        $command->setServer('serverRef');
        $command->setName('imageName');
        $command->setStatus('imageStatus');
        $command->setChangesSince('dateTime');
        $command->setType('type');
        $command->setMarker('markerID');
        $command->setLimit('int');

        $command->prepare();

        //Check method and resource
        $this->assertEquals('http://192.168.4.100:8774/v2/tenantid/images/detail?server=serverRef&name=imageName&status=imageStatus&changes-since=dateTime&type=type&marker=markerID&limit=int', $command->getRequest()->getUrl());
        $this->assertEquals('GET', $command->getRequest()->getMethod());

        //Check for authentication header
        $this->assertTrue($command->getRequest()->hasHeader('X-Auth-Token'));

        //Check parameters

        $this->client->execute($command);

        $result = $command->getResult();
        $this->assertTrue(is_array($result));

        $this->assertTrue(array_key_exists('images', $result));

    }
}

?>


text:
package util;

import io.MainPanel;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;

public class DatabaseBaseFnc {
	static Connection conn = null;
	final static String dbName = "Movies.db";

	/**
	 * Check if database table exists
	 * @return True if Movies table exists, false otherwise
	 */
	public static boolean pingDb() {
		try {
			conn = connect(conn);
			DatabaseMetaData md = conn.getMetaData();
            ResultSet rs = md.getTables(null, null, "MOVIES", null);
            conn.close();
            return rs.next();

		} catch(SQLException e) {
			e.printStackTrace();
		}
		return false;  //should never reach

	}
	/**
	 * Connect to database
	 */
    public static Connection connect(Connection conn){
        try{
        	Class.forName("org.sqlite.JDBC");
            conn = DriverManager.getConnection("jdbc:sqlite:" + dbName);
            return conn;
        }catch(ClassNotFoundException | SQLException e){
            e.printStackTrace();
            return null;
        }

    }
    /**
     * Initializes database table.  Create stmt
     * @return true if table created sucessfully
     */
    public static boolean initDb() {
    	//creates file for db
    	FileComs.createDbFile();
    	try {
        	conn = connect(conn);
    		Statement stmt = conn.createStatement();
    		String sqlInit = "CREATE TABLE MOVIES (ID INTEGER PRIMARY KEY AUTOINCREMENT, TITLE TEXT NOT NULL, RELEASE TEXT NOT NULL, GENRE TEXT NOT NULL, PLOT TEXT NOT NULL, LENGTH INT NOT NULL, RATING INT NOT NULL, LOCATION TEXT NOT NULL, TIMESTAMP DATETIME DEFAULT CURRENT_TIMESTAMP, SERIES_INFO TEXT NOT NULL, DISPLAY_INFO TEXT NOT NULL, IS_EPISODE BOOLEAN NOT NULL);";
    		stmt.executeUpdate(sqlInit);
    		stmt.close();
    		conn.close();
    		return true;
    	} catch(SQLException e) {
    		e.printStackTrace();
    		return false;
    	}
    }

    /**
     * Stores movie data in database
     * @param arr: movie information Title,Plot,Length,Release date,genre,imbd rating
     * @return
     */

    public static boolean storeMovie(Movie movie) {
    	String s = "INSERT into MOVIES "
				+ "(TITLE, RELEASE, GENRE, PLOT, LENGTH, RATING, LOCATION, SERIES_INFO, DISPLAY_INFO, IS_EPISODE) "
				+ " VALUES (?,?,?,?,?,?,?,?,?,?);"; //need to work
    	try {
			conn = connect(conn);
			PreparedStatement stmt = conn.prepareStatement(s);
			stmt.setString(1, movie.getTitle());
			stmt.setString(2, movie.getRelease());
			stmt.setString(3, movie.getGenre());
			stmt.setString(4, movie.getPlot());
			stmt.setString(5, movie.getLength());
			stmt.setString(6, movie.getRating());
			stmt.setString(7, movie.getLocation());
			stmt.setString(8, movie.getSeriesInfo());
			stmt.setString(9, movie.getDisplayInfo());
			stmt.setBoolean(10, movie.getIsSeriesBool());
			stmt.executeUpdate();
			stmt.close();
			conn.close();
			return true;
		} catch(SQLException e) {
			e.printStackTrace();
			return false;
		}
    }

    /**
     * Gets movies or series based on boolean series
     * @param series - Gets eries titles?
     * @return
     */
	public static ArrayList<Title> getTitles(boolean isSeries) {
		//Decide what we're getting
		String s = "";
		if(!isSeries)
			s = "SELECT ID, DISPLAY_INFO from Movies where IS_EPISODE = 0 ORDER BY DISPLAY_INFO ASC";
		else
			s = "SELECT ID, DISPLAY_INFO from MOVIES where IS_EPISODE = 1 ORDER BY DISPLAY_INFO ASC";

		ArrayList <Title> titles = new ArrayList<Title>();

		try {
			conn = connect(conn);
			Statement stmt = conn.createStatement();
			ResultSet rs = stmt.executeQuery(s);
			while(rs.next())
				titles.add(new Title(rs.getInt(1), rs.getString(2)));
			rs.close();
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}
		return titles;
	}

	public static ArrayList<Title> getTitlesSearch(boolean isSeries, String actionCommand, String text) {
		String s ="";
		if(!isSeries)
			s = "SELECT ID, DISPLAY_INFO from Movies where IS_EPISODE = 0 AND "+actionCommand+" LIKE ? ORDER BY DISPLAY_INFO ASC";
		else
			s = "SELECT ID, DISPLAY_INFO from MOVIES where IS_EPISODE = 1 AND "+actionCommand+" LIKE ? ORDER BY DISPLAY_INFO ASC";

		ArrayList<Title> titles = new ArrayList<Title>();

		try {
			conn = connect(conn);
			PreparedStatement stmt = conn.prepareStatement(s);
			stmt.setString(1, "%" + text + "%");
			ResultSet rs = stmt.executeQuery();
			while(rs.next())
				titles.add(new Title(rs.getInt(1), rs.getString(2)));
			rs.close();
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}

		return titles;
	}

	/**
	 * Removes a row from the database
	 * @param id
	 */
	public static void removeMovieFromDB(int id) {
		String s = "DELETE from MOVIES where ID = " + id;
		try {
			conn = connect(conn);
			Statement stmt = conn.createStatement();
			stmt.executeUpdate(s);
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}
	}

	/**
	 * Gets most recent entry
	 * @return
	 */
	public static Title getRecent() {
		String s = "SELECT ID, TITLE from Movies ORDER BY TIMESTAMP DESC LIMIT 1";
		Title recent = null;
		try {
			conn = connect(conn);
			Statement stmt = conn.createStatement();
			ResultSet rs = stmt.executeQuery(s);
			while(rs.next()) {
				recent = new Title(rs.getInt(1), rs.getString(2));
			}
			rs.close();
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}
		return recent;
	}

	/**
	 * Gets a full row from the database
	 * TODO: Should return movie type would be better
	 * @param id
	 * @return
	 */
	public static Movie getRow(int id) {
		Movie movie = new Movie();
		String s = "SELECT TITLE, RELEASE, GENRE, LENGTH, RATING, PLOT from MOVIES where id = " + Integer.toString(id);
		try {
			conn = connect(conn);
			Statement stmt = conn.createStatement();
			ResultSet rs = stmt.executeQuery(s);
			while(rs.next()) {
				movie.setTitle(rs.getString(1));
				movie.setRelease(rs.getString(2));
				movie.setGenre(rs.getString(3));
				movie.setLength(rs.getString(4));
				movie.setRating(rs.getString(5));
				movie.setPlot(rs.getString(6));
			}
			rs.close();
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}
		return movie;
	}

	/**
	 * Gets path of file with id associated with it
	 * @param id
	 * @return
	 */
	public static String getPath(int id) {
		String fp = "";
		String s = "SELECT LOCATION from MOVIES where ID = " + Integer.toString(id);
		try {
			conn = connect(conn);
			Statement stmt = conn.createStatement();
			ResultSet rs = stmt.executeQuery(s);
			while(rs.next()) {
				fp = rs.getString(1);
			}
			rs.close();
			stmt.close();
			conn.close();
		} catch(SQLException e) {
			e.printStackTrace();
		}
		return fp;
	}
}


text:
function J = computeCost(X, y, theta)
%COMPUTECOST Compute cost for linear regression
%   J = COMPUTECOST(X, y, theta) computes the cost of using theta as the
%   parameter for linear regression to fit the data points in X and y

m = length(y);

% matrix X has size m*(n+1) 97*2
% matrix theta has size (n+1)*1 2*1
% find hypothesis h(x) = theta0 + theta1*x

predictions = X*theta;	%97*1

% predictions of hypothesis on all m examples
% predictions is a m*1 matrix or a row vector

sqrErrors = (predictions-y).^2;	% do element wise square

J = 1/(2*m) * sum(sqrErrors);

% =========================================================================

end


text:
[jvm]
Content
fun [Builder](-builder.md)(defaults: [InstanceOfCheck](../index.md))
fun [Builder](-builder.md)()


text:
Registro

	<form name="registrationForm" id="registrationForm" class="form-horizontal css-form"  ng-submit="registrationForm.$valid && createNewUser()" novalidate>
		<fieldset>

				Nombre *

					<input class="form-control" id="inputName" ng-model="user.name" placeholder="Nombre" required>

				Apellidos *

					<input class="form-control" id="inputLastName" ng-model="user.lastName" placeholder="Apellidos" required>

				País

					<input class="form-control" id="inputCountry" ng-model="user.country" placeholder="País">

				Edad

					<input class="form-control" id="inputAge" ng-model="user.age" type="number" placeholder="Edad">

				Avatar

					<input type="file" id="inputFile" multiple="">

				Email *

					<input class="form-control" id="inputEmail" ng-model="user.email" placeholder="Correo electrónico"  type="email" required>

				Usuario *

					<input class="form-control" id="inputUsername" ng-model="user.username" placeholder="Nombre de Usuario" required>

				Contraseña *

					<input type="password" class="form-control" id="inputPassword" ng-model="user.password" placeholder="Contraseña" required data-bv-identical-message="Las contraseñas no coinciden">

				Confirmar *

					<input type="password" class="form-control" id="inputCheckPassword" ng-model="user.checkpassword" placeholder="Introduzca de nuevo su contraseña" required data-bv-identical-message="Las contraseñas no coinciden">

				<button type="button" class="btn btn-default" onClick="window.history.back()">Cancelar</button>
				<button type="submit" value="Submit" class="btn btn-primary" >Confirmar</button>

		</fieldset>
	</form>


text:
package ucesoft.cbm.peripheral.controlport

import net.java.games.input.{Component, Controller, ControllerEnvironment}
import ucesoft.cbm.Log
import ucesoft.cbm.peripheral.controlport.Joysticks._

import java.util.Properties

object GamePadControlPort {
  private var controllers : Array[Controller] = Array()

  private def discoverControllers(): Unit = {
    val thread = new Thread {
      override def run(): Unit = {
        System.setProperty("jinput.loglevel","SEVERE")
        controllers = ControllerEnvironment.getDefaultEnvironment.getControllers
        Log.info(s"JInput controllers discovery terminated [${controllers.length}]")
      }
    }
    thread.start()
    thread.join(1000)
  }

  def getControllers(): Array[Controller] = controllers
}

class GamePadControlPort(configuration:Properties) extends ControlPort {
  private[this] var controller : Option[Controller] = None
  private[this] var controllerName = ""
  private[this] var controllerFireName = ""
  private[this] var xAxisComponent : Option[Component] = None
  private[this] var yAxisComponent : Option[Component] = None
  private[this] var fireComponent : Option[Component] = None
  private[this] val dirThreshold = 0.5f

  GamePadControlPort.discoverControllers()
  findController

  def findController() : Unit = {
    controllerName = configuration.getProperty(CONFIG_CONTROLLER_NAME)
    controllerFireName = configuration.getProperty(CONFIG_CONTROLLER_FIRE_BUTTON,"1")
    val controllers = GamePadControlPort.getControllers()
    controller = controllers find { c => c.getName == controllerName  && (c.getType == Controller.Type.GAMEPAD || c.getType == Controller.Type.STICK)}
    controller match {
      case None =>
      case Some(comp) =>
        xAxisComponent = Option(comp.getComponent(Component.Identifier.Axis.X))
        yAxisComponent = Option(comp.getComponent(Component.Identifier.Axis.Y))
        fireComponent = comp.getComponents find { c => c.getIdentifier.getName == controllerFireName}
        //println("Find controller: " + comp.getName + " " + xAxisComponent + " " + yAxisComponent + " " + fireComponent)
    }
  }

  protected def read : Int = {
    var mask = 0
    controller match {
      case None =>
        if (controllerName != null) findController
      case Some(c) if c.poll =>
        for(x <- xAxisComponent;
        	y <- yAxisComponent;
        	f <- fireComponent) {
          if (f.getPollData != 0.0f) mask |= 16
          val xData = x.getPollData
          if (xData < -dirThreshold) mask |= 4 // left
          else
          if (xData > dirThreshold) mask |= 8 // right
          val yData = y.getPollData
          if (yData < -dirThreshold) mask |= 1 // up
          else
          if (yData > dirThreshold) mask |= 2 // down
        }

      case _ =>
        // controller unplugged
        controller = None
    }
    ~mask & 0xFF
  }
}


text:
import tempfile
import gnupg
import pytest

from django.conf import settings
from snoop import emails, models, pgp

pytestmark = pytest.mark.skipif(not settings.SNOOP_GPG_BINARY,
    reason="SNOOP_GPG_BINARY not set")

PATH_HUSH_MAIL = 'eml-9-pgp/encrypted-hushmail-knockoff.eml'
HEIN_PRIVATE_KEY = 'eml-9-pgp/keys/hein-priv.gpg'
HEIN_PUBLIC_KEY = 'eml-9-pgp/keys/hein-pub.gpg'

@pytest.yield_fixture(autouse=True)
def patch_gpg_to_temp_dir(monkeypatch):
    with tempfile.TemporaryDirectory() as tmp:
        gpg = gnupg.GPG(gnupghome=tmp, gpgbinary='gpg')
        for path in [HEIN_PRIVATE_KEY, HEIN_PUBLIC_KEY]:
            with open(settings.SNOOP_ROOT + "/" + path, 'rb') as f:
                gpg.import_keys(f.read())
        monkeypatch.setattr(pgp, 'GPG', gpg)
        monkeypatch.setattr(settings, 'SNOOP_GPG_HOME', '/tmp')
        yield

def create_email_doc(path, collection):
    doc = models.Document(
        path=path,
        content_type='message/rfc822',
        collection=collection,
    )
    doc.save = lambda *a, **k: None
    return doc

def parse_email(path, document_collection):
    return emails.parse_email(create_email_doc(path, document_collection))

def open_email(path, document_collection):
    return emails.open_email(create_email_doc(path, document_collection))

def test_doc_flags(document_collection):
    doc = create_email_doc(PATH_HUSH_MAIL, document_collection)
    emails.parse_email(doc)
    assert doc.flags.get('pgp')

def test_header_data(document_collection):
    data = parse_email(PATH_HUSH_MAIL, document_collection)
    assert data['subject'] == "Fwd: test email"
    assert data['date'] == '2016-08-10T15:00:00'

def test_attachments(document_collection):
    data = parse_email(PATH_HUSH_MAIL, document_collection)
    attach = data['attachments']
    assert len(attach) == 6

    email = open_email(PATH_HUSH_MAIL, document_collection)
    assert email.pgp

    with email.open_part('3') as f:
        text = f.read().decode()
        assert "This is GPG v1 speaking!" in text
        assert "Sent from my Android piece of !@#%." in text


text:
package org.rlcommunity.environments.knightjoust;

/**
 *
 * @author Ioannis Partalas
 */
import java.net.URL;
import org.rlcommunity.rlglue.codec.types.Action;
import org.rlcommunity.rlglue.codec.types.Observation;
import org.rlcommunity.rlglue.codec.taskspec.*;
import org.rlcommunity.rlglue.codec.types.Reward_observation_terminal;

import rlVizLib.Environments.EnvironmentBase;
import org.rlcommunity.rlglue.codec.taskspec.ranges.DoubleRange;
import org.rlcommunity.rlglue.codec.taskspec.ranges.IntRange;

import java.util.Random;
import rlVizLib.general.ParameterHolder;
import rlVizLib.messaging.environmentShell.TaskSpecPayload;
import rlVizLib.messaging.interfaces.HasImageInterface;

public class KnightJoust extends EnvironmentBase implements HasImageInterface {

    final protected KnightJoustState theState;
    private Random rand = new Random();

    public KnightJoust(ParameterHolder P) {
        theState=new KnightJoustState();
    }

    public KnightJoust() {
        this(getDefaultParameters());
    }

    //The following code is just some technical stuff so that we can check agent/env
    //compatibility in RL-Viz, and so we can later easily add parameters if we like.
    public static ParameterHolder getDefaultParameters() {
        return new ParameterHolder();
    }

    public static TaskSpecPayload getTaskSpecPayload(ParameterHolder P) {
        KnightJoust theEnv = new KnightJoust(P);
        String taskSpecString = theEnv.makeTaskSpec().getStringRepresentation();
        return new TaskSpecPayload(taskSpecString, false, "");
    }

    private TaskSpec makeTaskSpec() {
        TaskSpecVRLGLUE3 theTaskSpecObject = new TaskSpecVRLGLUE3();
        theTaskSpecObject.setEpisodic();
        theTaskSpecObject.setDiscountFactor(1.0d);

        theTaskSpecObject.addContinuousObservation(new DoubleRange(0.0, 34.0));
        theTaskSpecObject.addContinuousObservation(new DoubleRange(0.0, 180.0));
        theTaskSpecObject.addContinuousObservation(new DoubleRange(0.0, 180.0));

        theTaskSpecObject.addDiscreteAction(new IntRange(0, 2));
        theTaskSpecObject.setRewardRange(new DoubleRange(0, 50));
        theTaskSpecObject.setExtra("EnvName:KnightJoust:" + this.getClass().getPackage().getImplementationVersion());

        String taskSpecString = theTaskSpecObject.toTaskSpec();
        TaskSpec.checkTaskSpec(taskSpecString);

        return new TaskSpec(theTaskSpecObject);

    }

    public String env_init() {
        return makeTaskSpec().getStringRepresentation();
    }

    public Observation env_start() {
        theState.reset();

        return makeObservation();
    }

    public Reward_observation_terminal env_step(Action theAction) {

        int a = theAction.intArray[0];

        if (a > 2 || a < 0) {
            System.err.println("Invalid action selected in KnightJoust: " + a);
            a = rand.nextInt(2);
        }

        theState.updateState(a);

        return makeRewardObservation(theState.getReward(), theState.inGoalRegion());
    }

    @Override
    protected Observation makeObservation() {
        Observation currentObs = new Observation(0, 3);

        currentObs.doubleArray[0] = theState.calcDistToOpponent();
        currentObs.doubleArray[1] = theState.calcAngleEast();
        currentObs.doubleArray[2] = theState.calcAngleWest();

        return currentObs;
    }

    public String env_message(String theMessage) {
        return "Knight Joust does not respond to messages.";

    }

    public void env_cleanup() {
    }

    public URL getImageURL() {
        return this.getClass().getResource("/images/splash.png");
    }
}


text:
local lfs = require"lfs"

function attrdir (path)
    for file in lfs.dir(path) do
        if file ~= "." and file ~= ".." then
            local f = path..'/'..file
            print ("\t "..f)
            local attr = lfs.attributes (f)
            assert (type(attr) == "table")
            if attr.mode == "directory" then
                attrdir (f)
            else
                for name, value in pairs(attr) do
                    print (name, value)
                end
            end
        end
    end
end

attrdir (".")


text:
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Text;
using System.Windows.Forms;

namespace Unlocker
{
    public partial class FormSettings : Form
    {
        private Schedule schedule = null;

        public FormSettings()
        {
            InitializeComponent();
            schedule = new Schedule(this.timerTask);
        }

        private void comboBoxTime_SelectedIndexChanged(object sender, EventArgs e)
        {

        }

        /// <summary>
        /// 初始化默认候选
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void FormSettings_Load(object sender, EventArgs e)
        {
            this.comboBoxTime.SelectedIndex = 4;
        }

        /// <summary>
        /// 确定
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void buttonOk_Click(object sender, EventArgs e)
        {
            doTask();
            hideSettingsWindow();
        }

        private void doTask()
        {
            int min = this.comboBoxTime.SelectedIndex + 1;
            schedule.doTask(min);
        }
        /// <summary>
        /// 隐藏窗口,显示tips提示
        /// </summary>
        private void hideSettingsWindow()
        {
            //this.Hide();
            this.notifyIconUnlocker.Visible = true;//在通知区显示Form的Icon
            this.notifyIconUnlocker.ShowBalloonTip(1000, "运行提示", "防锁屏已经开始运行", ToolTipIcon.Info);
            this.WindowState = FormWindowState.Minimized;
            this.ShowInTaskbar = false;//使Form不在任务栏上显示
        }

        /// <summary>
        /// 显示设置界面
        /// </summary>
        private void showSettingsWindow()
        {
            this.WindowState = FormWindowState.Normal;
            this.notifyIconUnlocker.Visible = false;
            this.ShowInTaskbar = true;
        }

        /// <summary>
        /// 点击关闭按钮
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void FormSettings_FormClosing(object sender, FormClosingEventArgs e)
        {
            doTask();
            hideSettingsWindow();
        }

        /// <summary>
        /// 任务栏右键退出
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void toolStripMenuItemQuit_Click(object sender, EventArgs e)
        {
            this.schedule.stopTask();
            this.notifyIconUnlocker.Visible = false;
            this.Close();
        }

        /// <summary>
        /// 双击任务栏图标,出现设置界面
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void notifyIconUnlocker_DoubleClick(object sender, EventArgs e)
        {
            showSettingsWindow();
        }

        /// <summary>
        /// 任务栏右键设置
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void toolStripMenuItemSettings_Click(object sender, EventArgs e)
        {
            showSettingsWindow();
        }

        /// <summary>
        /// 任务栏右键关于
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void toolStripMenuItemAbout_Click(object sender, EventArgs e)
        {
            new FormAbout().ShowDialog();
        }
    }
}