text:
package org.springframework.richclient.form.binding.jide;

import com.jidesoft.swing.CheckBoxListWithSelectable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.binding.convert.ConversionExecutor;
import org.springframework.binding.form.FormModel;
import org.springframework.richclient.form.binding.swing.AbstractListBinding;
import org.springframework.util.Assert;

import javax.swing.*;
import java.awt.event.ItemEvent;
import java.awt.event.ItemListener;
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;

/**
 * A binding for the JIDE check box list with selectable component.
 *
 * This is a copy, paste and modification from the Spring RCP ListBinding class.
 *
 * @author Jonny Wray
 *
 */
public class CheckBoxListSelectableBinding extends AbstractListBinding {
	private static final Log logger = LogFactory.getLog(CheckBoxListSelectableBinding.class);
	private static final Object[] EMPTY_VALUES = new Object[0];

	private final PropertyChangeListener valueModelListener = new ValueModelListener();
	private final ItemListener checkBoxListener = new CheckBoxListener();

	private ConversionExecutor conversionExecutor;

	boolean selectingValues;

	public CheckBoxListSelectableBinding(CheckBoxListWithSelectable list,
			FormModel formModel, String formFieldPath, Class requiredSourceClass) {
		super(list, formModel, formFieldPath, requiredSourceClass);
	}

	public CheckBoxListWithSelectable getList() {
		return (CheckBoxListWithSelectable) getComponent();
	}

	public void setSelectionMode(int selectionMode) {
		Assert.isTrue(ListSelectionModel.SINGLE_SELECTION == selectionMode
				|| isPropertyConversionExecutorAvailable());
		getList().setSelectionMode(selectionMode);
	}

	public int getSelectionMode() {
		return getList().getSelectionMode();
	}

	/**
	 * Returns a conversion executor which converts a value of the given
	 * sourceType into the fieldType
	 *
	 * @see #getPropertyType()
	 */
	protected ConversionExecutor getPropertyConversionExecutor() {
		if (conversionExecutor == null) {
			conversionExecutor = getConversionService().getConversionExecutor(
					Object[].class, getPropertyType());
		}
		return conversionExecutor;
	}

	protected boolean isPropertyConversionExecutorAvailable() {
		try {
			getConversionService().getConversionExecutor(Object[].class,
					getPropertyType());
		} catch (IllegalArgumentException e) {
			return false;
		}
		return true;
	}

	protected void updateSelectedItemsFromSelectionModel() {
		if (getSelectionMode() == ListSelectionModel.SINGLE_SELECTION) {
			// JIDE CheckBoxList difference: get selected from CheckBoxList
			// method
			Object singleValue = getList().getSelectedObjects()[0];
			Class propertyType = getPropertyType();
			if (singleValue == null
					|| propertyType.isAssignableFrom(singleValue.getClass())) {
				getValueModel().setValueSilently(singleValue,
						valueModelListener);
			} else {
				getValueModel().setValueSilently(
						convertValue(singleValue, propertyType),
						valueModelListener);
			}
		} else {
			// JIDE CheckBoxList difference: get selected from CheckBoxList
			// method
			Object[] values = getList().getSelectedObjects();
			getValueModel().setValueSilently(convertSelectedValues(values),
					valueModelListener);
		}
	}

	/**
	 * Converts the given values into a type that matches the fieldType
	 *
	 * @param selectedValues
	 *            the selected values
	 * @return the value which can be assigned to the type of the field
	 */
	protected Object convertSelectedValues(Object[] selectedValues) {
		return getPropertyConversionExecutor().execute(selectedValues);
	}

	protected void doBindControl(ListModel bindingModel) {
		CheckBoxListWithSelectable list = getList();
		list.setModel(bindingModel);
		// JIDE CheckBoxListdifference: check box listener rather than list
		// selection
		list.addItemListener(checkBoxListener);
		getValueModel().addValueChangeListener(valueModelListener);
		if (!isPropertyConversionExecutorAvailable()
				&& getSelectionMode() != ListSelectionModel.SINGLE_SELECTION) {
			logger.warn("Selection mode for list field "
								+ getProperty()
								+ " forced to single selection."
								+ " If multiple selection is needed use a collection type (List, Collection, Object[])"
								+ " or provide a suitable converter to convert Object[] instances to property type "
								+ getPropertyType());

			setSelectionMode(ListSelectionModel.SINGLE_SELECTION);
		}
		updateSelectedItemsFromValueModel();
	}

	/**
	 * Updates the selection model with the selected values from the value
	 * model.
	 */
	protected void updateSelectedItemsFromValueModel() {
		Object value = getValue();
		Object[] selectedValues = EMPTY_VALUES;
		if (value != null) {
			selectedValues = (Object[]) convertValue(value, Object[].class);
		}

		// flag is used to avoid a round trip while we are selecting the values
		selectingValues = true;
		try {
			ListSelectionModel selectionModel = getList().getSelectionModel();
			selectionModel.setValueIsAdjusting(true);
			try {
				int[] valueIndexes = determineValueIndexes(selectedValues);
				int selectionMode = getSelectionMode();
				if (selectionMode == ListSelectionModel.SINGLE_SELECTION
						&& valueIndexes.length > 1) {
					getList().setSelectedIndex(valueIndexes[0]);
				} else {
					getList().setSelectedIndices(valueIndexes);
				}

				// update value model if selectedValues contain elements which
				// where not found in the list model
				// elements
				if (valueIndexes.length != selectedValues.length
						&& !isReadOnly()
						&& isEnabled()
						|| (selectionMode == ListSelectionModel.SINGLE_SELECTION && valueIndexes.length > 1)) {
					updateSelectedItemsFromSelectionModel();
				}
			} finally {
				selectionModel.setValueIsAdjusting(false);
			}
		} finally {
			selectingValues = false;
		}
	}

	/**
	 * @param values
	 * @return
	 */
	protected int[] determineValueIndexes(Object[] values) {
		int[] indexes = new int[values.length];
		if (values.length == 0)
			return indexes;

		Collection lookupValues = new HashSet(Arrays.asList(values));
		ListModel model = getList().getModel();
		int i = 0;
		for (int index = 0, size = model.getSize(); index < size
				&& !lookupValues.isEmpty(); index++) {
			if (lookupValues.remove(model.getElementAt(index))) {
				indexes[i++] = index;
			}
		}
		int[] result;
		if (i != values.length) {
			result = new int[i];
			System.arraycopy(indexes, 0, result, 0, i);
		} else {
			result = indexes;
		}
		return result;
	}

	public void setRenderer(ListCellRenderer renderer) {
		getList().setCellRenderer(renderer);
	}

	public ListCellRenderer getRenderer() {
		return getList().getCellRenderer();
	}

	// JIDE CheckBoxList difference. Change based on check boxes changing, not
	// list selection
	private class CheckBoxListener implements ItemListener {
		public void itemStateChanged(ItemEvent event) {
			if (!selectingValues) {
				updateSelectedItemsFromSelectionModel();
			}
		}
	}

	protected class ValueModelListener implements PropertyChangeListener {

		public void propertyChange(PropertyChangeEvent evt) {
			updateSelectedItemsFromValueModel();
		}

	}

	protected ListModel getDefaultModel() {
		return getList().getModel();
	}
}


text:
#ifndef _ALTOVA_INCLUDED_AIMXML_ALTOVA__ALTOVA_CtdType
#define _ALTOVA_INCLUDED_AIMXML_ALTOVA__ALTOVA_CtdType

namespace AIMXML
{

class CtdType : public TypeBase
{
public:
	AIMXML_EXPORT CtdType(xercesc::DOMNode* const& init);
	AIMXML_EXPORT CtdType(CtdType const& init);
	void operator=(CtdType const& other) { m_node = other.m_node; }
	static altova::meta::ComplexType StaticInfo() { return altova::meta::ComplexType(types + _altova_ti_altova_CtdType); }
	AIMXML_EXPORT void operator=(const string_type& value);
	AIMXML_EXPORT operator string_type();

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_ID, 0, 0> ID;	// ID CID

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_language, 0, 0> language;	// language CNMTOKEN

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_styleCode, 0, 0> styleCode;	// styleCode CNMTOKENS

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_abbr, 0, 0> abbr;	// abbr Cstring

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_axis, 0, 0> axis;	// axis Cstring

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_headers, 0, 0> headers;	// headers CIDREFS
	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_scope, 0, 4> scope;	// scope CscopeType

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_rowspan, 0, 0> rowspan;	// rowspan Cstring

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_colspan, 0, 0> colspan;	// colspan Cstring
	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_align, 0, 5> align;	// align CalignType4

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_char2, 0, 0> char2;	// char Cstring

	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_charoff, 0, 0> charoff;	// charoff Cstring
	MemberAttribute<string_type,_altova_mi_altova_CtdType_altova_valign, 0, 4> valign;	// valign CvalignType4
	MemberElement<CcontentType, _altova_mi_altova_CtdType_altova_content> content;
	struct content { typedef Iterator<CcontentType> iterator; };
	MemberElement<ClinkHtmlType, _altova_mi_altova_CtdType_altova_linkHtml> linkHtml;
	struct linkHtml { typedef Iterator<ClinkHtmlType> iterator; };
	MemberElement<xs::CstringType, _altova_mi_altova_CtdType_altova_sub> sub;
	struct sub { typedef Iterator<xs::CstringType> iterator; };
	MemberElement<xs::CstringType, _altova_mi_altova_CtdType_altova_sup> sup;
	struct sup { typedef Iterator<xs::CstringType> iterator; };
	MemberElement<CbrType, _altova_mi_altova_CtdType_altova_br> br;
	struct br { typedef Iterator<CbrType> iterator; };
	MemberElement<CfootnoteType, _altova_mi_altova_CtdType_altova_footnote> footnote;
	struct footnote { typedef Iterator<CfootnoteType> iterator; };
	MemberElement<CfootnoteRefType, _altova_mi_altova_CtdType_altova_footnoteRef> footnoteRef;
	struct footnoteRef { typedef Iterator<CfootnoteRefType> iterator; };
	MemberElement<CrenderMultiMediaType, _altova_mi_altova_CtdType_altova_renderMultiMedia> renderMultiMedia;
	struct renderMultiMedia { typedef Iterator<CrenderMultiMediaType> iterator; };
	MemberElement<CparagraphType, _altova_mi_altova_CtdType_altova_paragraph> paragraph;
	struct paragraph { typedef Iterator<CparagraphType> iterator; };
	MemberElement<ClistType, _altova_mi_altova_CtdType_altova_list> list;
	struct list { typedef Iterator<ClistType> iterator; };
	AIMXML_EXPORT void SetXsiType();
};

}	// namespace AIMXML

#endif // _ALTOVA_INCLUDED_AIMXML_ALTOVA__ALTOVA_CtdType

redpajama_set_name: RedPajamaGithub


text:
package org.shenit.tutorial.android.serialization;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.widget.TextView;

import org.shenit.tutorial.android.R;
import org.shenit.tutorial.android.entities.SerializableBook;

/**
 * 用Serializable实现序列化示例的跳转目标页面.
 */
public class SerializableToActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_book_detail);
        //获取搜索条件里的参数
        Bundle bundle = getIntent().getExtras();

        SerializableBook book = (SerializableBook) bundle.getSerializable("record");

        TextView titleTextView = (TextView) findViewById(R.id.title);
        TextView authorView = (TextView) findViewById(R.id.author);
        TextView contentView = (TextView) findViewById(R.id.content);

        titleTextView.setText(book.title);
        authorView.setText(book.author);
        contentView.setText(book.content);
    }
}


text:
#include "ogr_mysql.h"
#include "cpl_conv.h"
#include "cpl_string.h"

CPL_CVSID("$Id: ogrmysqllayer.cpp 28375 2015-01-30 12:06:11Z rouault $");

/************************************************************************/
/*                           OGRMySQLLayer()                            */
/************************************************************************/

OGRMySQLLayer::OGRMySQLLayer()

{
    poDS = NULL;

    pszGeomColumn = NULL;
    pszGeomColumnTable = NULL;
    pszFIDColumn = NULL;
    pszQueryStatement = NULL;

    bHasFid = FALSE;
    pszFIDColumn = NULL;

    iNextShapeId = 0;
    nResultOffset = 0;

    poSRS = NULL;
    nSRSId = -2; // we haven't even queried the database for it yet.

    poFeatureDefn = NULL;

    hResultSet = NULL;
}

/************************************************************************/
/*                           ~OGRMySQLLayer()                           */
/************************************************************************/

OGRMySQLLayer::~OGRMySQLLayer()

{
    if( m_nFeaturesRead > 0 && poFeatureDefn != NULL )
    {
        CPLDebug( "MySQL", "%d features read on layer '%s'.",
                  (int) m_nFeaturesRead,
                  poFeatureDefn->GetName() );
    }

    ResetReading();

    CPLFree( pszGeomColumn );
    CPLFree( pszGeomColumnTable );
    CPLFree( pszFIDColumn );
    CPLFree( pszQueryStatement );

    if( poSRS != NULL )
        poSRS->Release();

    if( poFeatureDefn )
        poFeatureDefn->Release();
}

/************************************************************************/
/*                            ResetReading()                            */
/************************************************************************/

void OGRMySQLLayer::ResetReading()

{
    iNextShapeId = 0;

    if( hResultSet != NULL )
    {
        mysql_free_result( hResultSet );
        hResultSet = NULL;

        poDS->InterruptLongResult();
    }
}

/************************************************************************/
/*                           GetNextFeature()                           */
/************************************************************************/

OGRFeature *OGRMySQLLayer::GetNextFeature()

{

    for( ; TRUE; )
    {
        OGRFeature      *poFeature;

        poFeature = GetNextRawFeature();
        if( poFeature == NULL )
            return NULL;

        if( (m_poFilterGeom == NULL
            || FilterGeometry( poFeature->GetGeometryRef() ) )
            && (m_poAttrQuery == NULL
                || m_poAttrQuery->Evaluate( poFeature )) )
            return poFeature;

        delete poFeature;
    }
}
/************************************************************************/
/*                          RecordToFeature()                           */
/*                                                                      */
/*      Convert the indicated record of the current result set into     */
/*      a feature.                                                      */
/************************************************************************/

OGRFeature *OGRMySQLLayer::RecordToFeature( char **papszRow,
                                            unsigned long *panLengths )

{
    mysql_field_seek( hResultSet, 0 );

/* -------------------------------------------------------------------- */
/*      Create a feature from the current result.                       */
/* -------------------------------------------------------------------- */
    int         iField;
    OGRFeature *poFeature = new OGRFeature( poFeatureDefn );

    poFeature->SetFID( iNextShapeId );
    m_nFeaturesRead++;

/* ==================================================================== */
/*      Transfer all result fields we can.                              */
/* ==================================================================== */
    for( iField = 0;
         iField < (int) mysql_num_fields(hResultSet);
         iField++ )
    {
        int     iOGRField;
        MYSQL_FIELD *psMSField = mysql_fetch_field(hResultSet);

/* -------------------------------------------------------------------- */
/*      Handle FID.                                                     */
/* -------------------------------------------------------------------- */
        if( bHasFid && EQUAL(psMSField->name,pszFIDColumn) )
        {
            if( papszRow[iField] == NULL )
            {
                CPLError( CE_Failure, CPLE_AppDefined,
                          "NULL primary key in RecordToFeature()" );
                return NULL;
            }

            poFeature->SetFID( CPLAtoGIntBig(papszRow[iField]) );
        }

        if( papszRow[iField] == NULL )
        {
//            CPLDebug("MYSQL", "%s was null for %d", psMSField->name,
//                     iNextShapeId);
            continue;
        }

/* -------------------------------------------------------------------- */
/*      Handle MySQL geometry                                           */
/* -------------------------------------------------------------------- */
        if( pszGeomColumn && EQUAL(psMSField->name,pszGeomColumn))
        {
            OGRGeometry *poGeometry = NULL;

            // Geometry columns will have the first 4 bytes contain the SRID.
            OGRGeometryFactory::createFromWkb(
                ((GByte *)papszRow[iField]) + 4,
                NULL,
                &poGeometry,
                panLengths[iField] - 4 );

            if( poGeometry != NULL )
            {
                poGeometry->assignSpatialReference( GetSpatialRef() );
                poFeature->SetGeometryDirectly( poGeometry );
            }
            continue;
        }

/* -------------------------------------------------------------------- */
/*      Transfer regular data fields.                                   */
/* -------------------------------------------------------------------- */
        iOGRField = poFeatureDefn->GetFieldIndex(psMSField->name);
        if( iOGRField < 0 )
            continue;

        OGRFieldDefn *psFieldDefn = poFeatureDefn->GetFieldDefn( iOGRField );

        if( psFieldDefn->GetType() == OFTBinary )
        {
            poFeature->SetField( iOGRField, panLengths[iField],
                                 (GByte *) papszRow[iField] );
        }
        else
        {
            poFeature->SetField( iOGRField, papszRow[iField] );
        }
    }

    return poFeature;
}

/************************************************************************/
/*                         GetNextRawFeature()                          */
/************************************************************************/

OGRFeature *OGRMySQLLayer::GetNextRawFeature()

{
/* -------------------------------------------------------------------- */
/*      Do we need to establish an initial query?                       */
/* -------------------------------------------------------------------- */
    if( iNextShapeId == 0 && hResultSet == NULL )
    {
        CPLAssert( pszQueryStatement != NULL );

        poDS->RequestLongResult( this );

        if( mysql_query( poDS->GetConn(), pszQueryStatement ) )
        {
            poDS->ReportError( pszQueryStatement );
            return NULL;
        }

        hResultSet = mysql_use_result( poDS->GetConn() );
        if( hResultSet == NULL )
        {
            poDS->ReportError( "mysql_use_result() failed on query." );
            return FALSE;
        }
    }

/* -------------------------------------------------------------------- */
/*      Fetch next record.                                              */
/* -------------------------------------------------------------------- */
    char **papszRow;
    unsigned long *panLengths;

    papszRow = mysql_fetch_row( hResultSet );
    if( papszRow == NULL )
    {
        ResetReading();
        return NULL;
    }

    panLengths = mysql_fetch_lengths( hResultSet );

/* -------------------------------------------------------------------- */
/*      Process record.                                                 */
/* -------------------------------------------------------------------- */
    OGRFeature *poFeature = RecordToFeature( papszRow, panLengths );

    iNextShapeId++;

    return poFeature;
}

/************************************************************************/
/*                             GetFeature()                             */
/*                                                                      */
/*      Note that we actually override this in OGRMySQLTableLayer.      */
/************************************************************************/

OGRFeature *OGRMySQLLayer::GetFeature( GIntBig nFeatureId )

{
    return OGRLayer::GetFeature( nFeatureId );
}

/************************************************************************/
/*                            GetFIDColumn()                            */
/************************************************************************/

const char *OGRMySQLLayer::GetFIDColumn()

{
    if( pszFIDColumn != NULL )
        return pszFIDColumn;
    else
        return "";
}

/************************************************************************/
/*                         GetGeometryColumn()                          */
/************************************************************************/

const char *OGRMySQLLayer::GetGeometryColumn()

{
    if( pszGeomColumn != NULL )
        return pszGeomColumn;
    else
        return "";
}

/************************************************************************/
/*                         FetchSRSId()                                 */
/************************************************************************/

int OGRMySQLLayer::FetchSRSId()
{
	CPLString        osCommand;
    char           **papszRow;

    if( hResultSet != NULL )
        mysql_free_result( hResultSet );
		hResultSet = NULL;

    osCommand.Printf(
             "SELECT srid FROM geometry_columns "
             "WHERE f_table_name = '%s'",
             pszGeomColumnTable );

    if( !mysql_query( poDS->GetConn(), osCommand ) )
        hResultSet = mysql_store_result( poDS->GetConn() );

    papszRow = NULL;
    if( hResultSet != NULL )
        papszRow = mysql_fetch_row( hResultSet );

    if( papszRow != NULL && papszRow[0] != NULL )
    {
        nSRSId = atoi(papszRow[0]);
    }

    // make sure to free our results
    if( hResultSet != NULL )
        mysql_free_result( hResultSet );
		hResultSet = NULL;

	return nSRSId;
}

/************************************************************************/
/*                           GetSpatialRef()                            */
/************************************************************************/

OGRSpatialReference *OGRMySQLLayer::GetSpatialRef()

{

    if( poSRS == NULL && nSRSId > -1 )
    {
        poSRS = poDS->FetchSRS( nSRSId );
        if( poSRS != NULL )
            poSRS->Reference();
        else
            nSRSId = -1;
    }

    return poSRS;

}


text:
package org.apache.spark.sql.execution.command.v1

import org.apache.spark.sql.Row
import org.apache.spark.sql.execution.command

/**
 * This base suite contains unified tests for the `DESCRIBE NAMESPACE` command that checks V1
 * table catalogs. The tests that cannot run for all V1 catalogs are located in more
 * specific test suites:
 *
 *   - V1 In-Memory catalog: `org.apache.spark.sql.execution.command.v1.DescribeNamespaceSuite`
 *   - V1 Hive External catalog:
 *     `org.apache.spark.sql.hive.execution.command.DescribeNamespaceSuite`
 */
trait DescribeNamespaceSuiteBase extends command.DescribeNamespaceSuiteBase
    with command.TestsV1AndV2Commands {
  override def notFoundMsgPrefix: String = "Database"

  test("basic") {
    val ns = "db1"
    withNamespace(ns) {
      sql(s"CREATE NAMESPACE $ns")

      val result = sql(s"DESCRIBE NAMESPACE EXTENDED $ns")
        .toDF("key", "value")
        .where("key not like 'Owner%'") // filter for consistency with in-memory catalog
        .collect()

      val namePrefix = if (conf.useV1Command) "Database" else "Namespace"
      assert(result.length == 4)
      assert(result(0) === Row(s"$namePrefix Name", ns))
      assert(result(1) === Row("Comment", ""))
      // Check only the key for "Location" since its value depends on warehouse path, etc.
      assert(result(2).getString(0) === "Location")
      assert(result(3) === Row("Properties", ""))
    }
  }
}

/**
 * The class contains tests for the `DESCRIBE NAMESPACE` command to check V1 In-Memory
 * table catalog.
 */
class DescribeNamespaceSuite extends DescribeNamespaceSuiteBase with CommandSuiteBase {
  override def commandVersion: String = super[DescribeNamespaceSuiteBase].commandVersion
}


text:
begin
  require 'nested_form/builder_mixin'

  module NestedForm
    class TwitterBootstrapBuilder < ::BootstrapForms::FormBuilder
      include ::NestedForm::BuilderMixin
    end
  end

  module BootstrapForms
    module Helpers
      module NestedFormHelper
        def bootstrap_nested_form_for(*args, &block)
          options = args.extract_options!.reverse_merge(:builder => NestedForm::TwitterBootstrapBuilder)
          form_for(*(args << options), &block) << after_nested_form_callbacks
        end
      end
    end
  end
rescue LoadError => e
  module BootstrapForms
    module Helpers
      module NestedFormHelper
        def bootstrap_nested_form_for(*args, &block)
          raise 'nested_form was not found. Is it in your Gemfile?'
        end
      end
    end
  end
end


text:
package producer

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"

	"github.com/betchi/tracer"
	logger "github.com/betchi/zapper"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/pkg/errors"
	"github.com/swagchat/chat-api/config"
	scpb "github.com/swagchat/protobuf/protoc-gen-go"
)

var KafkaConsumer *kafka.Consumer

type kafkaProvider struct {
	ctx context.Context
}

func (kp kafkaProvider) PublishMessage(rtmEvent *scpb.EventData) error {
	span := tracer.StartSpan(kp.ctx, "PublishMessage", "producer")
	defer tracer.Finish(span)

	cfg := config.Config()
	buffer := new(bytes.Buffer)
	json.NewEncoder(buffer).Encode(rtmEvent)

	p, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers": fmt.Sprintf("%s:%s", cfg.Producer.Kafka.Host, cfg.Producer.Kafka.Port),
	})
	if err != nil {
		err = errors.Wrap(err, "Kafka create producer failure")
		logger.Error(err.Error())
		tracer.SetError(span, err)
	}

	// Delivery report handler for produced messages
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					err = fmt.Errorf("Delivery failed: %v", ev.TopicPartition)
					logger.Error(err.Error())
					tracer.SetError(span, err)
				} else {
					logger.Info(fmt.Sprintf("Delivered message to %v\n", ev.TopicPartition))
				}
			}
		}
	}()

	// Produce messages to topic (asynchronously)
	topic := cfg.Producer.Kafka.Topic
	// for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} {
	p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          buffer.Bytes(),
	}, nil)
	// }

	// Wait for message deliveries
	p.Flush(15 * 1000)

	return nil
}


text:
<?xml version="1.0" encoding="utf-8"?><!--
 Copyright 2014 The Android Open Source Project

 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.
-->

<LinearLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:tools="http://schemas.android.com/tools" android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:background="#bb7700"
    android:orientation="horizontal">

    <com.example.android.tflitecamerademo.AutoFitTextureView
      android:id="@+id/texture"
      android:layout_width="0dp"
      android:layout_height="match_parent"
      android:layout_weight=".8"/>

    <LinearLayout
      android:layout_width="0dp"
      android:layout_height="match_parent"
      android:layout_weight=".2"
      android:orientation="vertical">

        <ImageView
            android:id="@+id/logoview"
            android:layout_width="wrap_content"
            android:layout_height="47dp"
            android:scaleType="centerInside"
            android:src="@drawable/logo"/>

        <TextView
            android:id="@+id/text"
            android:layout_width="match_parent"
            android:layout_height="160dp"
            android:paddingTop="20dp"
            android:textColor="#FFF"
            android:textSize="20sp"
            android:textStyle="bold"/>
        <LinearLayout
            android:id="@+id/modelLayout"
            android:layout_width="match_parent"
            android:layout_height="150dp"
            android:orientation="vertical">

            <TextView
                android:id="@+id/textView"
                android:layout_width="match_parent"
                android:layout_height="20dp"
                android:text="@string/modelLabel"
                android:textAlignment="center"
                android:textColor="@android:color/white"/>

            <ListView
                android:id="@+id/model"
                android:layout_width="match_parent"
                android:layout_height="wrap_content">

            </ListView>
        </LinearLayout>
        <LinearLayout
            android:id="@+id/deviceLayout"
            android:layout_width="match_parent"
            android:layout_height="150dp"
            android:orientation="vertical">

            <TextView
                android:id="@+id/textView2"
                android:layout_width="match_parent"
                android:layout_height="20dp"
                android:text="@string/deviceLabel"
                android:textAlignment="center"
                android:textColor="@android:color/white"/>

            <ListView
                android:id="@+id/device"
                android:layout_width="match_parent"
                android:layout_height="wrap_content"/>

        </LinearLayout>

        <LinearLayout
            android:layout_width="match_parent"
            android:layout_height="wrap_content"
        >

            <TextView
                android:layout_width="wrap_content"
                android:layout_height="wrap_content"
                android:gravity="center"
                android:text="Threads"
                android:textAlignment="center"
                android:textColor="@android:color/white"/>

            <NumberPicker
                android:id="@+id/np"
                android:layout_width="match_parent"
                android:layout_height="wrap_content"
                android:layout_marginLeft="10dp"
                android:theme="@style/AppTheme.Picker"
                android:visibility="visible"/>

        </LinearLayout>

    </LinearLayout>
</LinearLayout>


text:
package com.zimbra.common.soap;

import org.dom4j.Namespace;
import org.dom4j.QName;

public final class HsmConstants {
    public static final String NAMESPACE_STR = AdminConstants.NAMESPACE_STR;
    public static final Namespace NAMESPACE = Namespace.get(NAMESPACE_STR);

    public static final String E_HSM_REQUEST = "HsmRequest";
    public static final String E_HSM_RESPONSE = "HsmResponse";

    public static final String E_GET_HSM_STATUS_REQUEST = "GetHsmStatusRequest";
    public static final String E_GET_HSM_STATUS_RESPONSE = "GetHsmStatusResponse";

    public static final String E_ABORT_HSM_REQUEST = "AbortHsmRequest";
    public static final String E_ABORT_HSM_RESPONSE = "AbortHsmResponse";

    public static final String E_MOVE_BLOBS_REQUEST = "MoveBlobsRequest";
    public static final String E_MOVE_BLOBS_RESPONSE = "MoveBlobsResponse";

    public static final String E_GET_APPLIANCE_HSM_FS_REQUEST = "GetApplianceHSMFSRequest";
    public static final String E_GET_APPLIANCE_HSM_FS_RESPONSE = "GetApplianceHSMFSResponse";

    public static final QName HSM_REQUEST = QName.get(E_HSM_REQUEST, NAMESPACE);
    public static final QName HSM_RESPONSE = QName.get(E_HSM_RESPONSE, NAMESPACE);

    public static final QName GET_HSM_STATUS_REQUEST = QName.get(E_GET_HSM_STATUS_REQUEST, NAMESPACE);
    public static final QName GET_HSM_STATUS_RESPONSE = QName.get(E_GET_HSM_STATUS_RESPONSE, NAMESPACE);

    public static final QName ABORT_HSM_REQUEST = QName.get(E_ABORT_HSM_REQUEST, NAMESPACE);
    public static final QName ABORT_HSM_RESPONSE = QName.get(E_ABORT_HSM_RESPONSE, NAMESPACE);

    public static final QName MOVE_BLOBS_REQUEST = QName.get(E_MOVE_BLOBS_REQUEST, NAMESPACE);
    public static final QName MOVE_BLOBS_RESPONSE = QName.get(E_MOVE_BLOBS_RESPONSE, NAMESPACE);

    public static final QName GET_APPLIANCE_HSM_FS_REQUEST = QName.get(E_GET_APPLIANCE_HSM_FS_REQUEST, NAMESPACE);
    public static final QName GET_APPLIANCE_HSM_FS_RESPONSE = QName.get(E_GET_APPLIANCE_HSM_FS_RESPONSE, NAMESPACE);

    public static final String A_START_DATE = "startDate";
    public static final String A_END_DATE = "endDate";
    public static final String A_RUNNING = "running";
    public static final String A_ABORTED = "aborted";
    public static final String A_WAS_ABORTED = "wasAborted";
    public static final String A_ABORTING = "aborting";
    public static final String A_NUM_MOVED = "numMoved";
    public static final String A_ERROR = "error";
    public static final String A_NUM_BLOBS_MOVED = "numBlobsMoved";
    public static final String A_NUM_MAILBOXES = "numMailboxes";
    public static final String A_TOTAL_MAILBOXES = "totalMailboxes";
    public static final String A_DEST_VOLUME_ID = "destVolumeId";
    public static final String A_SOURCE_VOLUME_IDS = "sourceVolumeIds";
    public static final String A_MAX_BYTES = "maxBytes";
    public static final String A_NUM_BYTES_MOVED = "numBytesMoved";
    public static final String A_QUERY = "query";

    // GetApplianceHSMFS
    public static final String E_FS = "fs";
    public static final String A_SIZE = "size";
}


text:
import { curry, Curried2Result } from '../utils';
import contains from './contains';
import append from './append';

/**
 * @name addToList
 * @method
 * @memberof Frampton.List
 * @param {Array} xs  Array to add object to
 * @param {*}   obj Object to add to array
 * @returns {Array} A new array with the object added
 */
export default curry(<T>(xs: Array<T>, obj: T): Array<T> => {
  return <Array<T>>(
    (!contains(xs, obj)) ?
      append(xs, obj) :
      xs
  );
});


text:
namespace chromeos {
namespace secure_channel {

ConnectionManager::ConnectionManager() = default;
ConnectionManager::~ConnectionManager() = default;

void ConnectionManager::AddObserver(Observer* observer) {
  observer_list_.AddObserver(observer);
}

void ConnectionManager::RemoveObserver(Observer* observer) {
  observer_list_.RemoveObserver(observer);
}

void ConnectionManager::NotifyStatusChanged() {
  for (auto& observer : observer_list_)
    observer.OnConnectionStatusChanged();
}

void ConnectionManager::NotifyMessageReceived(const std::string& payload) {
  for (auto& observer : observer_list_)
    observer.OnMessageReceived(payload);
}

std::ostream& operator<<(std::ostream& stream,
                         ConnectionManager::Status status) {
  switch (status) {
    case ConnectionManager::Status::kConnecting:
      stream << "[Connecting]";
      break;
    case ConnectionManager::Status::kConnected:
      stream << "[Connected]";
      break;
    case ConnectionManager::Status::kDisconnected:
      stream << "[Disconnected]";
      break;
  }
  return stream;
}

}  // namespace secure_channel
}  // namespace chromeos


text:
import sys
from collections.abc import Iterator
from typing import cast, overload, Any, Callable, List, Optional, TYPE_CHECKING, Union

from py4j.java_gateway import java_import, JavaObject

from pyspark.sql.column import _to_seq
from pyspark.sql.readwriter import OptionUtils, to_str
from pyspark.sql.streaming.query import StreamingQuery
from pyspark.sql.types import Row, StructType
from pyspark.sql.utils import ForeachBatchFunction

if TYPE_CHECKING:
    from pyspark.sql.session import SparkSession
    from pyspark.sql._typing import SupportsProcess, OptionalPrimitiveType
    from pyspark.sql.dataframe import DataFrame

__all__ = ["DataStreamReader", "DataStreamWriter"]

class DataStreamReader(OptionUtils):
    """
    Interface used to load a streaming :class:`DataFrame <pyspark.sql.DataFrame>` from external
    storage systems (e.g. file systems, key-value stores, etc).
    Use :attr:`SparkSession.readStream <pyspark.sql.SparkSession.readStream>` to access this.

    .. versionadded:: 2.0.0

    Notes
    -----
    This API is evolving.

    Examples
    --------
    >>> spark.readStream
    <pyspark.sql.streaming.readwriter.DataStreamReader object ...>

    The example below uses Rate source that generates rows continously.
    After that, we operate a modulo by 3, and then writes the stream out to the console.
    The streaming query stops in 3 seconds.

    >>> import time
    >>> df = spark.readStream.format("rate").load()
    >>> df = df.selectExpr("value % 3 as v")
    >>> q = df.writeStream.format("console").start()
    >>> time.sleep(3)
    >>> q.stop()
    """

    def __init__(self, spark: "SparkSession") -> None:
        self._jreader = spark._jsparkSession.readStream()
        self._spark = spark

    def _df(self, jdf: JavaObject) -> "DataFrame":
        from pyspark.sql.dataframe import DataFrame

        return DataFrame(jdf, self._spark)

    def format(self, source: str) -> "DataStreamReader":
        """Specifies the input data source format.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        source : str
            name of the data source, e.g. 'json', 'parquet'.

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> spark.readStream.format("text")
        <pyspark.sql.streaming.readwriter.DataStreamReader object ...>

        This API allows to configure other sources to read. The example below writes a small text
        file, and reads it back via Text source.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary text file to read it.
        ...     spark.createDataFrame(
        ...         [("hello",), ("this",)]).write.mode("overwrite").format("text").save(d)
        ...
        ...     # Start a streaming query to read the text file.
        ...     q = spark.readStream.format("text").load(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._jreader = self._jreader.format(source)
        return self

    def schema(self, schema: Union[StructType, str]) -> "DataStreamReader":
        """Specifies the input schema.

        Some data sources (e.g. JSON) can infer the input schema automatically from data.
        By specifying the schema here, the underlying data source can skip the schema
        inference step, and thus speed up data loading.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        schema : :class:`pyspark.sql.types.StructType` or str
            a :class:`pyspark.sql.types.StructType` object or a DDL-formatted string
            (For example ``col0 INT, col1 DOUBLE``).

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> from pyspark.sql.types import StructField, StructType, StringType
        >>> spark.readStream.schema(StructType([StructField("data", StringType(), True)]))
        <pyspark.sql.streaming.readwriter.DataStreamReader object ...>
        >>> spark.readStream.schema("col0 INT, col1 DOUBLE")
        <pyspark.sql.streaming.readwriter.DataStreamReader object ...>

        The example below specifies a different schema to CSV file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Start a streaming query to read the CSV file.
        ...     spark.readStream.schema("col0 INT, col1 STRING").format("csv").load(d).printSchema()
        root
         |-- col0: integer (nullable = true)
         |-- col1: string (nullable = true)
        """
        from pyspark.sql import SparkSession

        spark = SparkSession._getActiveSessionOrCreate()
        if isinstance(schema, StructType):
            jschema = spark._jsparkSession.parseDataType(schema.json())
            self._jreader = self._jreader.schema(jschema)
        elif isinstance(schema, str):
            self._jreader = self._jreader.schema(schema)
        else:
            raise TypeError("schema should be StructType or string")
        return self

    def option(self, key: str, value: "OptionalPrimitiveType") -> "DataStreamReader":
        """Adds an input option for the underlying data source.

        .. versionadded:: 2.0.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> spark.readStream.option("x", 1)
        <pyspark.sql.streaming.readwriter.DataStreamReader object ...>

        The example below specifies 'rowsPerSecond' option to Rate source in order to generate
        10 rows every second.

        >>> import time
        >>> q = spark.readStream.format(
        ...     "rate").option("rowsPerSecond", 10).load().writeStream.format("console").start()
        >>> time.sleep(3)
        >>> q.stop()
        """
        self._jreader = self._jreader.option(key, to_str(value))
        return self

    def options(self, **options: "OptionalPrimitiveType") -> "DataStreamReader":
        """Adds input options for the underlying data source.

        .. versionadded:: 2.0.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> spark.readStream.options(x="1", y=2)
        <pyspark.sql.streaming.readwriter.DataStreamReader object ...>

        The example below specifies 'rowsPerSecond' and 'numPartitions' options to
        Rate source in order to generate 10 rows with 10 partitions every second.

        >>> import time
        >>> q = spark.readStream.format("rate").options(
        ...    rowsPerSecond=10, numPartitions=10
        ... ).load().writeStream.format("console").start()
        >>> time.sleep(3)
        >>> q.stop()
        """
        for k in options:
            self._jreader = self._jreader.option(k, to_str(options[k]))
        return self

    def load(
        self,
        path: Optional[str] = None,
        format: Optional[str] = None,
        schema: Optional[Union[StructType, str]] = None,
        **options: "OptionalPrimitiveType",
    ) -> "DataFrame":
        """Loads a data stream from a data source and returns it as a
        :class:`DataFrame <pyspark.sql.DataFrame>`.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        path : str, optional
            optional string for file-system backed data sources.
        format : str, optional
            optional string for format of the data source. Default to 'parquet'.
        schema : :class:`pyspark.sql.types.StructType` or str, optional
            optional :class:`pyspark.sql.types.StructType` for the input schema
            or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).
        **options : dict
            all other string options

        Notes
        -----
        This API is evolving.

        Examples
        --------
        Load a data stream from a temporary JSON file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary JSON file to read it.
        ...     spark.createDataFrame(
        ...         [(100, "Hyukjin Kwon"),], ["age", "name"]
        ...     ).write.mode("overwrite").format("json").save(d)
        ...
        ...     # Start a streaming query to read the JSON file.
        ...     q = spark.readStream.schema(
        ...         "age INT, name STRING"
        ...     ).format("json").load(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        if format is not None:
            self.format(format)
        if schema is not None:
            self.schema(schema)
        self.options(**options)
        if path is not None:
            if type(path) != str or len(path.strip()) == 0:
                raise ValueError(
                    "If the path is provided for stream, it needs to be a "
                    + "non-empty string. List of paths are not supported."
                )
            return self._df(self._jreader.load(path))
        else:
            return self._df(self._jreader.load())

    def json(
        self,
        path: str,
        schema: Optional[Union[StructType, str]] = None,
        primitivesAsString: Optional[Union[bool, str]] = None,
        prefersDecimal: Optional[Union[bool, str]] = None,
        allowComments: Optional[Union[bool, str]] = None,
        allowUnquotedFieldNames: Optional[Union[bool, str]] = None,
        allowSingleQuotes: Optional[Union[bool, str]] = None,
        allowNumericLeadingZero: Optional[Union[bool, str]] = None,
        allowBackslashEscapingAnyCharacter: Optional[Union[bool, str]] = None,
        mode: Optional[str] = None,
        columnNameOfCorruptRecord: Optional[str] = None,
        dateFormat: Optional[str] = None,
        timestampFormat: Optional[str] = None,
        multiLine: Optional[Union[bool, str]] = None,
        allowUnquotedControlChars: Optional[Union[bool, str]] = None,
        lineSep: Optional[str] = None,
        locale: Optional[str] = None,
        dropFieldIfAllNull: Optional[Union[bool, str]] = None,
        encoding: Optional[str] = None,
        pathGlobFilter: Optional[Union[bool, str]] = None,
        recursiveFileLookup: Optional[Union[bool, str]] = None,
        allowNonNumericNumbers: Optional[Union[bool, str]] = None,
    ) -> "DataFrame":
        """
        Loads a JSON file stream and returns the results as a :class:`DataFrame`.

        `JSON Lines <http://jsonlines.org/>`_ (newline-delimited JSON) is supported by default.
        For JSON (one record per file), set the ``multiLine`` parameter to ``true``.

        If the ``schema`` parameter is not specified, this function goes
        through the input once to determine the input schema.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        path : str
            string represents path to the JSON dataset,
            or RDD of Strings storing JSON objects.
        schema : :class:`pyspark.sql.types.StructType` or str, optional
            an optional :class:`pyspark.sql.types.StructType` for the input schema
            or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).

        Other Parameters
        ----------------
        Extra options
            For the extra options, refer to
            `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-json.html#data-source-option>`_
            in the version you use.

            .. # noqa

        Notes
        -----
        This API is evolving.

        Examples
        --------
        Load a data stream from a temporary JSON file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary JSON file to read it.
        ...     spark.createDataFrame(
        ...         [(100, "Hyukjin Kwon"),], ["age", "name"]
        ...     ).write.mode("overwrite").format("json").save(d)
        ...
        ...     # Start a streaming query to read the JSON file.
        ...     q = spark.readStream.schema(
        ...         "age INT, name STRING"
        ...     ).json(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._set_opts(
            schema=schema,
            primitivesAsString=primitivesAsString,
            prefersDecimal=prefersDecimal,
            allowComments=allowComments,
            allowUnquotedFieldNames=allowUnquotedFieldNames,
            allowSingleQuotes=allowSingleQuotes,
            allowNumericLeadingZero=allowNumericLeadingZero,
            allowBackslashEscapingAnyCharacter=allowBackslashEscapingAnyCharacter,
            mode=mode,
            columnNameOfCorruptRecord=columnNameOfCorruptRecord,
            dateFormat=dateFormat,
            timestampFormat=timestampFormat,
            multiLine=multiLine,
            allowUnquotedControlChars=allowUnquotedControlChars,
            lineSep=lineSep,
            locale=locale,
            dropFieldIfAllNull=dropFieldIfAllNull,
            encoding=encoding,
            pathGlobFilter=pathGlobFilter,
            recursiveFileLookup=recursiveFileLookup,
            allowNonNumericNumbers=allowNonNumericNumbers,
        )
        if isinstance(path, str):
            return self._df(self._jreader.json(path))
        else:
            raise TypeError("path can be only a single string")

    def orc(
        self,
        path: str,
        mergeSchema: Optional[bool] = None,
        pathGlobFilter: Optional[Union[bool, str]] = None,
        recursiveFileLookup: Optional[Union[bool, str]] = None,
    ) -> "DataFrame":
        """Loads a ORC file stream, returning the result as a :class:`DataFrame`.

        .. versionadded:: 2.3.0

        Other Parameters
        ----------------
        Extra options
            For the extra options, refer to
            `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-orc.html#data-source-option>`_
            in the version you use.

            .. # noqa

        Examples
        --------
        Load a data stream from a temporary ORC file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary ORC file to read it.
        ...     spark.range(10).write.mode("overwrite").format("orc").save(d)
        ...
        ...     # Start a streaming query to read the ORC file.
        ...     q = spark.readStream.schema("id LONG").orc(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._set_opts(
            mergeSchema=mergeSchema,
            pathGlobFilter=pathGlobFilter,
            recursiveFileLookup=recursiveFileLookup,
        )
        if isinstance(path, str):
            return self._df(self._jreader.orc(path))
        else:
            raise TypeError("path can be only a single string")

    def parquet(
        self,
        path: str,
        mergeSchema: Optional[bool] = None,
        pathGlobFilter: Optional[Union[bool, str]] = None,
        recursiveFileLookup: Optional[Union[bool, str]] = None,
        datetimeRebaseMode: Optional[Union[bool, str]] = None,
        int96RebaseMode: Optional[Union[bool, str]] = None,
    ) -> "DataFrame":
        """
        Loads a Parquet file stream, returning the result as a :class:`DataFrame`.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        path : str
            the path in any Hadoop supported file system

        Other Parameters
        ----------------
        Extra options
            For the extra options, refer to
            `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#data-source-option>`_.
            in the version you use.

            .. # noqa

        Examples
        --------
        Load a data stream from a temporary Parquet file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary Parquet file to read it.
        ...     spark.range(10).write.mode("overwrite").format("parquet").save(d)
        ...
        ...     # Start a streaming query to read the Parquet file.
        ...     q = spark.readStream.schema(
        ...         "id LONG").parquet(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._set_opts(
            mergeSchema=mergeSchema,
            pathGlobFilter=pathGlobFilter,
            recursiveFileLookup=recursiveFileLookup,
            datetimeRebaseMode=datetimeRebaseMode,
            int96RebaseMode=int96RebaseMode,
        )
        if isinstance(path, str):
            return self._df(self._jreader.parquet(path))
        else:
            raise TypeError("path can be only a single string")

    def text(
        self,
        path: str,
        wholetext: bool = False,
        lineSep: Optional[str] = None,
        pathGlobFilter: Optional[Union[bool, str]] = None,
        recursiveFileLookup: Optional[Union[bool, str]] = None,
    ) -> "DataFrame":
        """
        Loads a text file stream and returns a :class:`DataFrame` whose schema starts with a
        string column named "value", and followed by partitioned columns if there
        are any.
        The text files must be encoded as UTF-8.

        By default, each line in the text file is a new row in the resulting DataFrame.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        path : str or list
            string, or list of strings, for input path(s).

        Other Parameters
        ----------------
        Extra options
            For the extra options, refer to
            `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-text.html#data-source-option>`_
            in the version you use.

            .. # noqa

        Notes
        -----
        This API is evolving.

        Examples
        --------
        Load a data stream from a temporary text file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary text file to read it.
        ...     spark.createDataFrame(
        ...         [("hello",), ("this",)]).write.mode("overwrite").format("text").save(d)
        ...
        ...     # Start a streaming query to read the text file.
        ...     q = spark.readStream.text(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._set_opts(
            wholetext=wholetext,
            lineSep=lineSep,
            pathGlobFilter=pathGlobFilter,
            recursiveFileLookup=recursiveFileLookup,
        )
        if isinstance(path, str):
            return self._df(self._jreader.text(path))
        else:
            raise TypeError("path can be only a single string")

    def csv(
        self,
        path: str,
        schema: Optional[Union[StructType, str]] = None,
        sep: Optional[str] = None,
        encoding: Optional[str] = None,
        quote: Optional[str] = None,
        escape: Optional[str] = None,
        comment: Optional[str] = None,
        header: Optional[Union[bool, str]] = None,
        inferSchema: Optional[Union[bool, str]] = None,
        ignoreLeadingWhiteSpace: Optional[Union[bool, str]] = None,
        ignoreTrailingWhiteSpace: Optional[Union[bool, str]] = None,
        nullValue: Optional[str] = None,
        nanValue: Optional[str] = None,
        positiveInf: Optional[str] = None,
        negativeInf: Optional[str] = None,
        dateFormat: Optional[str] = None,
        timestampFormat: Optional[str] = None,
        maxColumns: Optional[Union[int, str]] = None,
        maxCharsPerColumn: Optional[Union[int, str]] = None,
        maxMalformedLogPerPartition: Optional[Union[int, str]] = None,
        mode: Optional[str] = None,
        columnNameOfCorruptRecord: Optional[str] = None,
        multiLine: Optional[Union[bool, str]] = None,
        charToEscapeQuoteEscaping: Optional[Union[bool, str]] = None,
        enforceSchema: Optional[Union[bool, str]] = None,
        emptyValue: Optional[str] = None,
        locale: Optional[str] = None,
        lineSep: Optional[str] = None,
        pathGlobFilter: Optional[Union[bool, str]] = None,
        recursiveFileLookup: Optional[Union[bool, str]] = None,
        unescapedQuoteHandling: Optional[str] = None,
    ) -> "DataFrame":
        r"""Loads a CSV file stream and returns the result as a :class:`DataFrame`.

        This function will go through the input once to determine the input schema if
        ``inferSchema`` is enabled. To avoid going through the entire data once, disable
        ``inferSchema`` option or specify the schema explicitly using ``schema``.

        Parameters
        ----------
        path : str or list
            string, or list of strings, for input path(s).
        schema : :class:`pyspark.sql.types.StructType` or str, optional
            an optional :class:`pyspark.sql.types.StructType` for the input schema
            or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).

        .. versionadded:: 2.0.0

        Other Parameters
        ----------------
        Extra options
            For the extra options, refer to
            `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-csv.html#data-source-option>`_
            in the version you use.

            .. # noqa

        Notes
        -----
        This API is evolving.

        Examples
        --------
        Load a data stream from a temporary CSV file.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Write a temporary text file to read it.
        ...     spark.createDataFrame([(1, "2"),]).write.mode("overwrite").format("csv").save(d)
        ...
        ...     # Start a streaming query to read the CSV file.
        ...     q = spark.readStream.schema(
        ...         "col0 INT, col1 STRING"
        ...     ).format("csv").load(d).writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q.stop()
        """
        self._set_opts(
            schema=schema,
            sep=sep,
            encoding=encoding,
            quote=quote,
            escape=escape,
            comment=comment,
            header=header,
            inferSchema=inferSchema,
            ignoreLeadingWhiteSpace=ignoreLeadingWhiteSpace,
            ignoreTrailingWhiteSpace=ignoreTrailingWhiteSpace,
            nullValue=nullValue,
            nanValue=nanValue,
            positiveInf=positiveInf,
            negativeInf=negativeInf,
            dateFormat=dateFormat,
            timestampFormat=timestampFormat,
            maxColumns=maxColumns,
            maxCharsPerColumn=maxCharsPerColumn,
            maxMalformedLogPerPartition=maxMalformedLogPerPartition,
            mode=mode,
            columnNameOfCorruptRecord=columnNameOfCorruptRecord,
            multiLine=multiLine,
            charToEscapeQuoteEscaping=charToEscapeQuoteEscaping,
            enforceSchema=enforceSchema,
            emptyValue=emptyValue,
            locale=locale,
            lineSep=lineSep,
            pathGlobFilter=pathGlobFilter,
            recursiveFileLookup=recursiveFileLookup,
            unescapedQuoteHandling=unescapedQuoteHandling,
        )
        if isinstance(path, str):
            return self._df(self._jreader.csv(path))
        else:
            raise TypeError("path can be only a single string")

    def table(self, tableName: str) -> "DataFrame":
        """Define a Streaming DataFrame on a Table. The DataSource corresponding to the table should
        support streaming mode.

        .. versionadded:: 3.1.0

        Parameters
        ----------
        tableName : str
            string, for the name of the table.

        Returns
        -------
        :class:`DataFrame`

        Notes
        -----
        This API is evolving.

        Examples
        --------
        Load a data stream from a table.

        >>> import tempfile
        >>> import time
        >>> _ = spark.sql("DROP TABLE IF EXISTS my_table")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Create a table with Rate source.
        ...     q1 = spark.readStream.format("rate").load().writeStream.toTable(
        ...         "my_table", checkpointLocation=d)
        ...
        ...     # Read the table back and print out in the console.
        ...     q2 = spark.readStream.table("my_table").writeStream.format("console").start()
        ...     time.sleep(3)
        ...     q1.stop()
        ...     q2.stop()
        ...     _ = spark.sql("DROP TABLE my_table")
        """
        if isinstance(tableName, str):
            return self._df(self._jreader.table(tableName))
        else:
            raise TypeError("tableName can be only a single string")

class DataStreamWriter:
    """
    Interface used to write a streaming :class:`DataFrame <pyspark.sql.DataFrame>` to external
    storage systems (e.g. file systems, key-value stores, etc).
    Use :attr:`DataFrame.writeStream <pyspark.sql.DataFrame.writeStream>`
    to access this.

    .. versionadded:: 2.0.0

    Notes
    -----
    This API is evolving.

    Examples
    --------
    The example below uses Rate source that generates rows continously.
    After that, we operate a modulo by 3, and then writes the stream out to the console.
    The streaming query stops in 3 seconds.

    >>> import time
    >>> df = spark.readStream.format("rate").load()
    >>> df = df.selectExpr("value % 3 as v")
    >>> q = df.writeStream.format("console").start()
    >>> time.sleep(3)
    >>> q.stop()
    """

    def __init__(self, df: "DataFrame") -> None:
        self._df = df
        self._spark = df.sparkSession
        self._jwrite = df._jdf.writeStream()

    def _sq(self, jsq: JavaObject) -> StreamingQuery:
        return StreamingQuery(jsq)

    def outputMode(self, outputMode: str) -> "DataStreamWriter":
        """Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.

        .. versionadded:: 2.0.0

        Options include:

        * `append`: Only the new rows in the streaming DataFrame/Dataset will be written to
           the sink
        * `complete`: All the rows in the streaming DataFrame/Dataset will be written to the sink
           every time these are some updates
        * `update`: only the rows that were updated in the streaming DataFrame/Dataset will be
           written to the sink every time there are some updates. If the query doesn't contain
           aggregations, it will be equivalent to `append` mode.

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()
        >>> df.writeStream.outputMode('append')
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        The example below uses Complete mode that the entire aggregated counts are printed out.

        >>> import time
        >>> df = spark.readStream.format("rate").option("rowsPerSecond", 10).load()
        >>> df = df.groupby().count()
        >>> q = df.writeStream.outputMode("complete").format("console").start()
        >>> time.sleep(3)
        >>> q.stop()
        """
        if not outputMode or type(outputMode) != str or len(outputMode.strip()) == 0:
            raise ValueError("The output mode must be a non-empty string. Got: %s" % outputMode)
        self._jwrite = self._jwrite.outputMode(outputMode)
        return self

    def format(self, source: str) -> "DataStreamWriter":
        """Specifies the underlying output data source.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        source : str
            string, name of the data source, which for now can be 'parquet'.

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()
        >>> df.writeStream.format("text")
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        This API allows to configure the source to write. The example below writes a CSV
        file from Rate source in a streaming manner.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d, tempfile.TemporaryDirectory() as cp:
        ...     df = spark.readStream.format("rate").load()
        ...     q = df.writeStream.format("csv").option("checkpointLocation", cp).start(d)
        ...     time.sleep(5)
        ...     q.stop()
        ...     spark.read.schema("timestamp TIMESTAMP, value STRING").csv(d).show()
        +...---------+-----+
        |...timestamp|value|
        +...---------+-----+
        ...
        """
        self._jwrite = self._jwrite.format(source)
        return self

    def option(self, key: str, value: "OptionalPrimitiveType") -> "DataStreamWriter":
        """Adds an output option for the underlying data source.

        .. versionadded:: 2.0.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()
        >>> df.writeStream.option("x", 1)
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        The example below specifies 'numRows' option to Console source in order to print
        3 rows for every batch.

        >>> import time
        >>> q = spark.readStream.format(
        ...     "rate").option("rowsPerSecond", 10).load().writeStream.format(
        ...         "console").option("numRows", 3).start()
        >>> time.sleep(3)
        >>> q.stop()
        """
        self._jwrite = self._jwrite.option(key, to_str(value))
        return self

    def options(self, **options: "OptionalPrimitiveType") -> "DataStreamWriter":
        """Adds output options for the underlying data source.

        .. versionadded:: 2.0.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()
        >>> df.writeStream.option("x", 1)
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        The example below specifies 'numRows' and 'truncate' options to Console source in order
        to print 3 rows for every batch without truncating the results.

        >>> import time
        >>> q = spark.readStream.format(
        ...     "rate").option("rowsPerSecond", 10).load().writeStream.format(
        ...         "console").options(numRows=3, truncate=False).start()
        >>> time.sleep(3)
        >>> q.stop()
        """
        for k in options:
            self._jwrite = self._jwrite.option(k, to_str(options[k]))
        return self

    @overload
    def partitionBy(self, *cols: str) -> "DataStreamWriter":
        ...

    @overload
    def partitionBy(self, __cols: List[str]) -> "DataStreamWriter":
        ...

    def partitionBy(self, *cols: str) -> "DataStreamWriter":  # type: ignore[misc]
        """Partitions the output by the given columns on the file system.

        If specified, the output is laid out on the file system similar
        to Hive's partitioning scheme.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        cols : str or list
            name of columns

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()
        >>> df.writeStream.partitionBy("value")
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        Partition-by timestamp column from Rate source.

        >>> import tempfile
        >>> import time
        >>> with tempfile.TemporaryDirectory() as d, tempfile.TemporaryDirectory() as cp:
        ...     df = spark.readStream.format("rate").option("rowsPerSecond", 10).load()
        ...     q = df.writeStream.partitionBy(
        ...         "timestamp").format("parquet").option("checkpointLocation", cp).start(d)
        ...     time.sleep(5)
        ...     q.stop()
        ...     spark.read.schema(df.schema).parquet(d).show()
        +...---------+-----+
        |...timestamp|value|
        +...---------+-----+
        ...
        """
        if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
            cols = cols[0]
        self._jwrite = self._jwrite.partitionBy(_to_seq(self._spark._sc, cols))
        return self

    def queryName(self, queryName: str) -> "DataStreamWriter":
        """Specifies the name of the :class:`StreamingQuery` that can be started with
        :func:`start`. This name must be unique among all the currently active queries
        in the associated SparkSession.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        queryName : str
            unique name for the query

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> import time
        >>> df = spark.readStream.format("rate").load()
        >>> q = df.writeStream.queryName("streaming_query").format("console").start()
        >>> q.stop()
        >>> q.name
        'streaming_query'
        """
        if not queryName or type(queryName) != str or len(queryName.strip()) == 0:
            raise ValueError("The queryName must be a non-empty string. Got: %s" % queryName)
        self._jwrite = self._jwrite.queryName(queryName)
        return self

    @overload
    def trigger(self, *, processingTime: str) -> "DataStreamWriter":
        ...

    @overload
    def trigger(self, *, once: bool) -> "DataStreamWriter":
        ...

    @overload
    def trigger(self, *, continuous: str) -> "DataStreamWriter":
        ...

    @overload
    def trigger(self, *, availableNow: bool) -> "DataStreamWriter":
        ...

    def trigger(
        self,
        *,
        processingTime: Optional[str] = None,
        once: Optional[bool] = None,
        continuous: Optional[str] = None,
        availableNow: Optional[bool] = None,
    ) -> "DataStreamWriter":
        """Set the trigger for the stream query. If this is not set it will run the query as fast
        as possible, which is equivalent to setting the trigger to ``processingTime='0 seconds'``.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        processingTime : str, optional
            a processing time interval as a string, e.g. '5 seconds', '1 minute'.
            Set a trigger that runs a microbatch query periodically based on the
            processing time. Only one trigger can be set.
        once : bool, optional
            if set to True, set a trigger that processes only one batch of data in a
            streaming query then terminates the query. Only one trigger can be set.
        continuous : str, optional
            a time interval as a string, e.g. '5 seconds', '1 minute'.
            Set a trigger that runs a continuous query with a given checkpoint
            interval. Only one trigger can be set.
        availableNow : bool, optional
            if set to True, set a trigger that processes all available data in multiple
            batches then terminates the query. Only one trigger can be set.

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()

        Trigger the query for execution every 5 seconds

        >>> df.writeStream.trigger(processingTime='5 seconds')
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        Trigger the query for execution every 5 seconds

        >>> df.writeStream.trigger(continuous='5 seconds')
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>

        Trigger the query for reading all available data with multiple batches

        >>> df.writeStream.trigger(availableNow=True)
        <pyspark.sql.streaming.readwriter.DataStreamWriter object ...>
        """
        params = [processingTime, once, continuous, availableNow]

        if params.count(None) == 4:
            raise ValueError("No trigger provided")
        elif params.count(None) < 3:
            raise ValueError("Multiple triggers not allowed.")

        jTrigger = None
        assert self._spark._sc._jvm is not None
        if processingTime is not None:
            if type(processingTime) != str or len(processingTime.strip()) == 0:
                raise ValueError(
                    "Value for processingTime must be a non empty string. Got: %s" % processingTime
                )
            interval = processingTime.strip()
            jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.ProcessingTime(
                interval
            )

        elif once is not None:
            if once is not True:
                raise ValueError("Value for once must be True. Got: %s" % once)
            jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.Once()

        elif continuous is not None:
            if type(continuous) != str or len(continuous.strip()) == 0:
                raise ValueError(
                    "Value for continuous must be a non empty string. Got: %s" % continuous
                )
            interval = continuous.strip()
            jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.Continuous(
                interval
            )
        else:
            if availableNow is not True:
                raise ValueError("Value for availableNow must be True. Got: %s" % availableNow)
            jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.AvailableNow()

        self._jwrite = self._jwrite.trigger(jTrigger)
        return self

    @overload
    def foreach(self, f: Callable[[Row], None]) -> "DataStreamWriter":
        ...

    @overload
    def foreach(self, f: "SupportsProcess") -> "DataStreamWriter":
        ...

    def foreach(self, f: Union[Callable[[Row], None], "SupportsProcess"]) -> "DataStreamWriter":
        """
        Sets the output of the streaming query to be processed using the provided writer ``f``.
        This is often used to write the output of a streaming query to arbitrary storage systems.
        The processing logic can be specified in two ways.

        #. A **function** that takes a row as input.
            This is a simple way to express your processing logic. Note that this does
            not allow you to deduplicate generated data when failures cause reprocessing of
            some input data. That would require you to specify the processing logic in the next
            way.

        #. An **object** with a ``process`` method and optional ``open`` and ``close`` methods.
            The object can have the following methods.

            * ``open(partition_id, epoch_id)``: *Optional* method that initializes the processing
                (for example, open a connection, start a transaction, etc). Additionally, you can
                use the `partition_id` and `epoch_id` to deduplicate regenerated data
                (discussed later).

            * ``process(row)``: *Non-optional* method that processes each :class:`Row`.

            * ``close(error)``: *Optional* method that finalizes and cleans up (for example,
                close connection, commit transaction, etc.) after all rows have been processed.

            The object will be used by Spark in the following way.

            * A single copy of this object is responsible of all the data generated by a
                single task in a query. In other words, one instance is responsible for
                processing one partition of the data generated in a distributed manner.

            * This object must be serializable because each task will get a fresh
                serialized-deserialized copy of the provided object. Hence, it is strongly
                recommended that any initialization for writing data (e.g. opening a
                connection or starting a transaction) is done after the `open(...)`
                method has been called, which signifies that the task is ready to generate data.

            * The lifecycle of the methods are as follows.

                For each partition with ``partition_id``:

                ... For each batch/epoch of streaming data with ``epoch_id``:

                ....... Method ``open(partitionId, epochId)`` is called.

                ....... If ``open(...)`` returns true, for each row in the partition and
                        batch/epoch, method ``process(row)`` is called.

                ....... Method ``close(errorOrNull)`` is called with error (if any) seen while
                        processing rows.

            Important points to note:

            * The `partitionId` and `epochId` can be used to deduplicate generated data when
                failures cause reprocessing of some input data. This depends on the execution
                mode of the query. If the streaming query is being executed in the micro-batch
                mode, then every partition represented by a unique tuple (partition_id, epoch_id)
                is guaranteed to have the same data. Hence, (partition_id, epoch_id) can be used
                to deduplicate and/or transactionally commit data and achieve exactly-once
                guarantees. However, if the streaming query is being executed in the continuous
                mode, then this guarantee does not hold and therefore should not be used for
                deduplication.

            * The ``close()`` method (if exists) will be called if `open()` method exists and
                returns successfully (irrespective of the return value), except if the Python
                crashes in the middle.

        .. versionadded:: 2.4.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> import time
        >>> df = spark.readStream.format("rate").load()

        Print every row using a function

        >>> def print_row(row):
        ...     print(row)
        ...
        >>> q = df.writeStream.foreach(print_row).start()
        >>> time.sleep(3)
        >>> q.stop()

        Print every row using a object with process() method

        >>> class RowPrinter:
        ...     def open(self, partition_id, epoch_id):
        ...         print("Opened %d, %d" % (partition_id, epoch_id))
        ...         return True
        ...
        ...     def process(self, row):
        ...         print(row)
        ...
        ...     def close(self, error):
        ...         print("Closed with error: %s" % str(error))
        ...
        >>> q = df.writeStream.foreach(print_row).start()
        >>> time.sleep(3)
        >>> q.stop()
        """

        from pyspark.rdd import _wrap_function
        from pyspark.serializers import CPickleSerializer, AutoBatchedSerializer
        from pyspark.taskcontext import TaskContext

        if callable(f):
            # The provided object is a callable function that is supposed to be called on each row.
            # Construct a function that takes an iterator and calls the provided function on each
            # row.
            def func_without_process(_: Any, iterator: Iterator) -> Iterator:
                for x in iterator:
                    f(x)  # type: ignore[operator]
                return iter([])

            func = func_without_process

        else:
            # The provided object is not a callable function. Then it is expected to have a
            # 'process(row)' method, and optional 'open(partition_id, epoch_id)' and
            # 'close(error)' methods.

            if not hasattr(f, "process"):
                raise AttributeError("Provided object does not have a 'process' method")

            if not callable(getattr(f, "process")):
                raise TypeError("Attribute 'process' in provided object is not callable")

            def doesMethodExist(method_name: str) -> bool:
                exists = hasattr(f, method_name)
                if exists and not callable(getattr(f, method_name)):
                    raise TypeError(
                        "Attribute '%s' in provided object is not callable" % method_name
                    )
                return exists

            open_exists = doesMethodExist("open")
            close_exists = doesMethodExist("close")

            def func_with_open_process_close(partition_id: Any, iterator: Iterator) -> Iterator:
                epoch_id = cast(TaskContext, TaskContext.get()).getLocalProperty(
                    "streaming.sql.batchId"
                )
                if epoch_id:
                    int_epoch_id = int(epoch_id)
                else:
                    raise RuntimeError("Could not get batch id from TaskContext")

                # Check if the data should be processed
                should_process = True
                if open_exists:
                    should_process = f.open(partition_id, int_epoch_id)  # type: ignore[union-attr]

                error = None

                try:
                    if should_process:
                        for x in iterator:
                            cast("SupportsProcess", f).process(x)
                except Exception as ex:
                    error = ex
                finally:
                    if close_exists:
                        f.close(error)  # type: ignore[union-attr]
                    if error:
                        raise error

                return iter([])

            func = func_with_open_process_close  # type: ignore[assignment]

        serializer = AutoBatchedSerializer(CPickleSerializer())
        wrapped_func = _wrap_function(self._spark._sc, func, serializer, serializer)
        assert self._spark._sc._jvm is not None
        jForeachWriter = (
            self._spark._sc._jvm.org.apache.spark.sql.execution.python.PythonForeachWriter(
                wrapped_func, self._df._jdf.schema()
            )
        )
        self._jwrite.foreach(jForeachWriter)
        return self

    def foreachBatch(self, func: Callable[["DataFrame", int], None]) -> "DataStreamWriter":
        """
        Sets the output of the streaming query to be processed using the provided
        function. This is supported only the in the micro-batch execution modes (that is, when the
        trigger is not continuous). In every micro-batch, the provided function will be called in
        every micro-batch with (i) the output rows as a DataFrame and (ii) the batch identifier.
        The batchId can be used deduplicate and transactionally write the output
        (that is, the provided Dataset) to external systems. The output DataFrame is guaranteed
        to exactly same for the same batchId (assuming all operations are deterministic in the
        query).

        .. versionadded:: 2.4.0

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> import time
        >>> df = spark.readStream.format("rate").load()
        >>> def func(batch_df, batch_id):
        ...     batch_df.collect()
        ...
        >>> q = df.writeStream.foreachBatch(func).start()
        >>> time.sleep(3)
        >>> q.stop()
        """

        from pyspark.java_gateway import ensure_callback_server_started

        gw = self._spark._sc._gateway
        assert gw is not None
        java_import(gw.jvm, "org.apache.spark.sql.execution.streaming.sources.*")

        wrapped_func = ForeachBatchFunction(self._spark, func)
        gw.jvm.PythonForeachBatchHelper.callForeachBatch(self._jwrite, wrapped_func)
        ensure_callback_server_started(gw)
        return self

    def start(
        self,
        path: Optional[str] = None,
        format: Optional[str] = None,
        outputMode: Optional[str] = None,
        partitionBy: Optional[Union[str, List[str]]] = None,
        queryName: Optional[str] = None,
        **options: "OptionalPrimitiveType",
    ) -> StreamingQuery:
        """Streams the contents of the :class:`DataFrame` to a data source.

        The data source is specified by the ``format`` and a set of ``options``.
        If ``format`` is not specified, the default data source configured by
        ``spark.sql.sources.default`` will be used.

        .. versionadded:: 2.0.0

        Parameters
        ----------
        path : str, optional
            the path in a Hadoop supported file system
        format : str, optional
            the format used to save
        outputMode : str, optional
            specifies how data of a streaming DataFrame/Dataset is written to a
            streaming sink.

            * `append`: Only the new rows in the streaming DataFrame/Dataset will be written to the
              sink
            * `complete`: All the rows in the streaming DataFrame/Dataset will be written to the
              sink every time these are some updates
            * `update`: only the rows that were updated in the streaming DataFrame/Dataset will be
              written to the sink every time there are some updates. If the query doesn't contain
              aggregations, it will be equivalent to `append` mode.
        partitionBy : str or list, optional
            names of partitioning columns
        queryName : str, optional
            unique name for the query
        **options : dict
            All other string options. You may want to provide a `checkpointLocation`
            for most streams, however it is not required for a `memory` stream.

        Notes
        -----
        This API is evolving.

        Examples
        --------
        >>> df = spark.readStream.format("rate").load()

        Basic example.

        >>> q = df.writeStream.format('memory').queryName('this_query').start()
        >>> q.isActive
        True
        >>> q.name
        'this_query'
        >>> q.stop()
        >>> q.isActive
        False

        Example with using other parameters with a trigger.

        >>> q = df.writeStream.trigger(processingTime='5 seconds').start(
        ...     queryName='that_query', outputMode="append", format='memory')
        >>> q.name
        'that_query'
        >>> q.isActive
        True
        >>> q.stop()
        """
        self.options(**options)
        if outputMode is not None:
            self.outputMode(outputMode)
        if partitionBy is not None:
            self.partitionBy(partitionBy)
        if format is not None:
            self.format(format)
        if queryName is not None:
            self.queryName(queryName)
        if path is None:
            return self._sq(self._jwrite.start())
        else:
            return self._sq(self._jwrite.start(path))

    def toTable(
        self,
        tableName: str,
        format: Optional[str] = None,
        outputMode: Optional[str] = None,
        partitionBy: Optional[Union[str, List[str]]] = None,
        queryName: Optional[str] = None,
        **options: "OptionalPrimitiveType",
    ) -> StreamingQuery:
        """
        Starts the execution of the streaming query, which will continually output results to the
        given table as new data arrives.

        The returned :class:`StreamingQuery` object can be used to interact with the stream.

        .. versionadded:: 3.1.0

        Parameters
        ----------
        tableName : str
            string, for the name of the table.
        format : str, optional
            the format used to save.
        outputMode : str, optional
            specifies how data of a streaming DataFrame/Dataset is written to a
            streaming sink.

            * `append`: Only the new rows in the streaming DataFrame/Dataset will be written to the
              sink
            * `complete`: All the rows in the streaming DataFrame/Dataset will be written to the
              sink every time these are some updates
            * `update`: only the rows that were updated in the streaming DataFrame/Dataset will be
              written to the sink every time there are some updates. If the query doesn't contain
              aggregations, it will be equivalent to `append` mode.
        partitionBy : str or list, optional
            names of partitioning columns
        queryName : str, optional
            unique name for the query
        **options : dict
            All other string options. You may want to provide a `checkpointLocation`.

        Notes
        -----
        This API is evolving.

        For v1 table, partitioning columns provided by `partitionBy` will be respected no matter
        the table exists or not. A new table will be created if the table not exists.

        For v2 table, `partitionBy` will be ignored if the table already exists. `partitionBy` will
        be respected only if the v2 table does not exist. Besides, the v2 table created by this API
        lacks some functionalities (e.g., customized properties, options, and serde info). If you
        need them, please create the v2 table manually before the execution to avoid creating a
        table with incomplete information.

        Examples
        --------
        Save a data stream to a table.

        >>> import tempfile
        >>> import time
        >>> _ = spark.sql("DROP TABLE IF EXISTS my_table2")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     # Create a table with Rate source.
        ...     q = spark.readStream.format("rate").option(
        ...         "rowsPerSecond", 10).load().writeStream.toTable(
        ...             "my_table2",
        ...             queryName='that_query',
        ...             outputMode="append",
        ...             format='parquet',
        ...             checkpointLocation=d)
        ...     time.sleep(3)
        ...     q.stop()
        ...     spark.read.table("my_table2").show()
        ...     _ = spark.sql("DROP TABLE my_table2")
        +...---------+-----+
        |...timestamp|value|
        +...---------+-----+
        ...
        """
        self.options(**options)
        if outputMode is not None:
            self.outputMode(outputMode)
        if partitionBy is not None:
            self.partitionBy(partitionBy)
        if format is not None:
            self.format(format)
        if queryName is not None:
            self.queryName(queryName)
        return self._sq(self._jwrite.toTable(tableName))

def _test() -> None:
    import doctest
    import os
    from pyspark.sql import SparkSession
    import pyspark.sql.streaming.readwriter

    os.chdir(os.environ["SPARK_HOME"])

    globs = pyspark.sql.streaming.readwriter.__dict__.copy()
    globs["spark"] = (
        SparkSession.builder.master("local[4]")
        .appName("sql.streaming.readwriter tests")
        .getOrCreate()
    )

    (failure_count, test_count) = doctest.testmod(
        pyspark.sql.streaming.readwriter,
        globs=globs,
        optionflags=doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE | doctest.REPORT_NDIFF,
    )
    globs["spark"].stop()

    if failure_count:
        sys.exit(-1)

if __name__ == "__main__":
    _test()


text:
Formas de pago

Podes hacer tu pago en efectivo, con tarjeta o a través de Rapipago/Pagofacil.

Efectivo

Transferencia bancaria a través del banco

Utilizando tu tarjeta de debido desde un cajero

Es instantáneo, y el que más recomendamos

Tarjetas

Utilizamos Mercado Pago, por lo cual aceptamos todas las tarjetas

Podes ver las promociones con las tarjetas desde acá

Es instantáneo

Rapipago / Pagofacil

Generas una factura y vas a pagar.

Tarda entre 2 hs a 48 hs laborales en acreditarse.


text:
"use strict";

class RedTop {
    constructor(){
        this.zones = [];
    }

    //Checks for unique AvailabilityZone name
    addAvailabilityZone(az) {
        var f = false;
        this.zones.forEach(function(zone){
            if (zone.getName() == az.getName()){
                f = true;
            }
        });
        if (!f) this.zones.push(az);
    }

    //Supports a string arg to remove by name or remove by index
    delAvailabilityZone(az) {
        if(typeof(az) == "string"){
            this.zones.forEach(function(z, i){
                if(z.getName() == az.getName())
                this.zones.slice(i, i+1);
            });
        }
        else if (typeof(az) == "number"){
            this.zones.slice(az, az+1);
        }
    }

    addSubnet(s, az){
        var f = false;
        this.zones.forEach(function(zone){

            if(az.getName() == zone.getName()){
                zone.subnets.forEach(function(sn){
                    if(sn.getNetId() == s.getNetId()) f = true;
                });

                if(!f) zone.addSubnet(s);
            }
        });

    }

    delSubnet(s, az){
        this.zones.forEach(function(zone){
            if(zone.getName() == az.getName()){
                zone.delSubnet(s);
            }
        });
    }

    addInstance(i, s, az){
        this.zones.forEach(function(zone){
            if(zone.getName() == az.getName()){
                zone.subnets.forEach(function(net){
                    if(net.getNetId() == s.getNetId()){
                        net.addInstance(i);
                    }
                });
            }
        });
    }

    delInstance(i, s, az){
        this.zones.forEach(function(zone){
            if(zone.getName() == az.getName()){
                zone.subnets.forEach(function(net){
                    if(net.getNetId() == s.getNetId()){
                        net.delInstance(i);
                    }
                });
            }
        });
    }

    addNode(n, i, s, az){
        this.zones.forEach(function(zone){
            if(zone.getName() == az.getName()){
                zone.subnets.forEach(function(net){
                    if(net.getNetId() == s.getNetId()){
                        net.instances.forEach(function(inst){
                            if(inst.getId() == i.getId()){
                                inst.addNode(n);
                            }
                        });
                    }
                });
            }
        });
    }

    delNode(n, i, s, az){
        this.zones.forEach(function(zone){
            if(zone.getName() == az.getName()){
                zone.subnets.forEach(function(net){
                    if(net.getNetId() == s.getNetId()){
                        net.instances.forEach(function(inst){
                            if(inst.getId() == i.getId()){
                                inst.delNode(n);
                            }
                        });
                    }
                });
            }
        });
    }

    getAvailabilityZones(){
        return this.zones;
    }

    getSubnets(az){
        if (az){
            this.zones.forEach(function(zone){
                if(zone.getName() == az.getName()){
                    return zone.getSubnets();
                }
            });
        }
        else {
            var subnets = [];
            this.zones.forEach(function(zone){
                zone.getSubnets().forEach(function(net){
                    subnets.push(net);
                });
            });
            return subnets;
        }
    }

    getInstances(s, az){

        //Only az is given
        if(arguments.length == 1){
            var instances = [];
            this.zones.forEach(function(zone){
                if(zone.getName() == az.getName()){
                    zone.getSubnets().forEach(function(net){
                        net.getInstances().forEach(function(inst){
                            instances.push(inst);
                        });
                    });
                }
            });
            return instances;
        }
        else if (arguments.length == 2){
            this.zones.forEach(function(zone){
                if(zone.getName() == az.getName()){
                    zone.getSubnets().forEach(function(net){
                        if(net.getId() == s.getId()){
                            return net.getInstances();
                        }
                    });
                }
            });
        }
        else{
            var instances = [];
            var nets = this.getSubnets();
            nets.forEach(function(net){
                net.getInstances().forEach(function(inst){
                    instances.push(inst);
                })
            });
            return instances;
        }
    }

    getNodes(i, sn, az){
        if(arguments.length == 3){
            //Get nodes at specified instance
        }
        else if (arguments.length == 2){
            //Get nodes in specified subnet
        }
        else if (arguments.length == 1){
            //get nodes in specified az
        }
        else{
            var nodes = [];
            var inst = this.getInstances();
            inst.forEach(function(instance){
                instance.getNodes().forEach(function(n){
                    nodes.push(n);
                });
            });

            return nodes;
        }
    }

    getSlaves(){
        var s = [];

        this.getNodes().forEach(function(node){
            if (node.getRole().toUpperCase() == "SLAVE")
                s.push(node);
        });

        return s;
    }

    getMasters(){
        var m = [];

        this.getNodes().forEach(function(node){
            if (node.getRole().toUpperCase() == "MASTER")
                m.push(node);
        });

        return m;
    }
};