Add example for python function

functions/python/README.md

@@ -0,0 +1,47 @@
+# python-function
+
+This is an example of how to use the python function.
+
+## Project Structure
+
+```
+python_function
+├── __init__.py
+├── custom_object_function.py
+└── pyserde
+    ├── __init__.py
+    └── serde.py
+```
+
+In python, if a folder contains `__init__` file, we call it a package, python_function and pyserde are both packages. Compressing `python_function` directory will generate `python_function.zip`.
+
+## How to use
+
+### Start pulsar standalone
+
+```
+docker run -d -it \
+    -p 6650:6650 \
+    -p 8080:8080 \
+    --name pulsar-standalone \
+    apachepulsar/pulsar-all:2.7.1 \
+    bin/pulsar standalone
+```
+
+### Start python function by use zip package
+```
+./bin/pulsar-admin functions create \
+    --tenant public   --namespace default   --name my_function \
+    --py /YOUR-PATH/python_function.zip \
+    --classname python_function.custom_object_function.CustomObjectFunction \
+    --custom-serde-inputs '{"input-topic-1":"python_function.pyserde.serde.CustomSerDe","input-topic-2":"python_function.pyserde.serde.CustomSerDe"}' \
+    --output-serde-classname python_function.pyserde.serde.CustomSerDe \
+    --output output-topic-3
+```
+Replace `YOUR-PATH` with your file path.
+
+### Run produce and consume
+
+```
+python test-producer-consumer.py
+```

functions/python/python_function/custom_object_function.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+
+from pulsar import Function, SerDe
+from python_function.pyserde.serde import MyObject
+
+# Function that deals with custom objects
+class CustomObjectFunction(Function):
+  def __init__(self):
+    pass
+
+  def process(self, input, context):
+    retval = MyObject()
+    retval.a = input.a + 11
+    retval.b = input.b + 24
+    return retval

functions/python/python_function/pyserde/serde.py

@@ -0,0 +1,20 @@
+from pulsar import SerDe
+
+class MyObject(object):
+  def __init__(self):
+    self.a = 0
+    self.b = 0
+
+class CustomSerDe(SerDe):
+  def __init__(self):
+    pass
+
+  def serialize(self, object):
+    return ("%d,%d" % (object.a, object.b)).encode('utf-8')
+
+  def deserialize(self, input_bytes):
+    split = str(input_bytes.decode()).split(',')
+    retval = MyObject()
+    retval.a = int(split[0])
+    retval.b = int(split[1])
+    return retval

functions/python/test-producer-consumer.py

@@ -0,0 +1,41 @@
+from pulsar import Client, SerDe
+
+class MyObject(object):
+  def __init__(self):
+    self.a = 0
+    self.b = 0
+
+class CustomSerDe(SerDe):
+  def __init__(self):
+    pass
+  def serialize(self, object):
+    return ("%d,%d" % (object.a, object.b)).encode('utf-8')
+  def deserialize(self, input_bytes):
+    split = str(input_bytes.decode()).split(',')
+    retval = MyObject()
+    retval.a = int(split[0])
+    retval.b = int(split[1])
+    return retval
+
+client = Client('pulsar://localhost:6650')
+producer1 = client.create_producer(topic='input-topic-1')
+producer2 = client.create_producer(topic='input-topic-2')
+consumer = client.subscribe(
+                  topic='output-topic-3',
+                  subscription_name='my-subscription')
+
+ser = CustomSerDe()
+for i in range(100):
+  dataObject = MyObject()
+  dataObject.a = i
+  dataObject.b = i + 100
+  d = ser.serialize(dataObject)
+  producer1.send(d)
+  producer2.send(d)
+  print('send msg "%d", "%d"' % (dataObject.a, dataObject.b))
+for i in range(100):
+    msg = consumer.receive()
+    consumer.acknowledge(msg)
+    ex = msg.value()
+    d = ser.deserialize(ex)
+    print('receive and ack msg "%d", "%d"' % (d.a, d.b))