查看 Application Integration 支援的連接器。

Cloud 函式工作

Cloud Function 工作可讓您透過整合服務設定及執行 Cloud Run 函式。Cloud Run 函式是 Cloud Functions 的進階版，提供事件導向的程式設計模型，並運用 Cloud Run 無伺服器平台，強化控制和擴充功能。Cloud Run 函式提供單一無伺服器解決方案，適用於所有工作負載類型。

Cloud Functions 工作支援下列版本的 Cloud Run 函式：

• Cloud Functions (第 1 代)
• 使用 Cloud Functions v2 API 建立的 Cloud Run 函式

如要詳細瞭解 Cloud Run functions 版本之間的差異，請參閱 Cloud Functions 比較指南。

注意：本文會交互使用 Cloud Run functions (第 1 代)、Cloud Functions (第 1 代) 和 Cloud Functions 等詞彙。

事前準備

請務必先在 Google Cloud 專案中執行下列工作，再設定 Cloud Function 工作。

1. 如要連線至 Cloud Function，請確認您已建立 OAuth 2.0 設定檔，或將使用者管理的服務帳戶附加至整合：
• 如果整合項目已附加服務帳戶，請將 Cloud Functions 叫用者 IAM 角色指派給該服務帳戶。

  如要瞭解如何將角色授予服務帳戶，請參閱「管理服務帳戶的存取權」一文。

• 「Cloud Function」工作僅支援「Google OIDC ID 權杖」類型的驗證設定檔。 使用已指派 Cloud Functions Invoker IAM 角色的服務帳戶，建立 Google OIDC ID 憑證類型的驗證設定檔。 如果 Cloud Function 工作不需要驗證，工作設定窗格中的「驗證設定檔」欄位可以留空。

如果整合服務同時設定了 OIDC ID 設定檔和使用者管理的服務帳戶，系統預設會使用 OIDC ID 設定檔進行驗證。如果未設定 OIDC ID 設定檔和使用者管理的服務帳戶，系統會使用預設服務帳戶 ([e-mail]) 呼叫 Cloud Function 工作。

2. 確認您 Google Cloud 專案中的 Application Integration 未設定 VPC Service Controls。
   警告：如果為專案中的 Application Integration 設定 VPC Service Controls，Cloud Functions 將無法運作或停止運作。 Google Cloud

設定 Cloud Functions 工作

如要在整合中設定 Cloud 函式工作，請按照下列步驟操作：

1. 在導覽選單中，按一下「整合」。

   這時「應用程式整合」使用者介面會開啟，並顯示可用的整合功能清單。

2. 選取現有的整合服務，或按一下「建立整合服務」。

   如要建立新的整合，請在建立對話方塊中輸入名稱和說明，然後按一下「建立」。

3. 從「Tasks」(任務) 下拉式選單中，按一下「Cloud Function」(雲端函式)，將其放置在整合服務編輯器中。
4. 按一下設計工具中的「Cloud Function」(雲端函式) 元素，開啟設定窗格，然後點選「Configure Cloud Function」(設定 Cloud Function)。
5. 如果系統提示您為服務帳戶授予權限，請按一下「授予」。

   Application Integration 會自動將必要權限授予服務帳戶。

6. 在「Cloud Function configuration」(Cloud Functions 設定) 窗格中，選取下列其中一個選項：
   • 連結現有函式：選取這個選項，將現有函式與整合服務建立關聯。您可以從整合服務連結 Cloud Functions (第 1 代) 和使用 Cloud Functions v2 API 建立的 Cloud Functions。
     • 在「Cloud Function 觸發網址」欄位中，輸入現有函式的觸發網址。

       網址應採用下列其中一種格式：

       # For Cloud Functions (1st gen)
       https://REGION_NAME-PROJECT_ID.cloudfunctions.net/FUNCTION_NAME

       # For Cloud Run functions created using the Cloud Functions v2 API
       https://FUNCTION_NAME-PROJECT_ID.REGION_NAME.run.app

     重要事項：請確認現有 Cloud Function 的來源檔案符合Cloud Function 範本中指定的格式。
   • 建立新函式：選取這個選項，為整合服務建立新函式。
     1. 在「Function Name」(函式名稱) 欄位中，輸入新 Cloud Run 函式的專屬名稱。
     2. 從「Region」(區域) 下拉式選單中，選取 Cloud Run 函式部署的區域。
     3. 從「函式版本」下拉式選單中，選取所需的 Cloud Run 函式版本：
        • Cloud Functions (第 1 代)：這是 Cloud Run 函式的舊版，舊稱 Cloud Functions (第 1 代)，使用 .cloudfunctions.net 端點格式。
        • Cloud Functions (最新一代)：這是最新版本的 Cloud Run 函式，使用 Cloud Functions v2 API 建立。這項服務以 Cloud Run 和 Eventarc 為基礎，支援延長要求逾時時間 (最多 60 分鐘)、提高並行處理量，並使用 .cloudfunctions.net 和 .run.app 端點格式。
        如要進一步瞭解這兩個版本的差異，請參閱「比較 Cloud Functions」。
7. 按一下「Save」(儲存)。

在 Application Integration 中設定 Cloud Functions 工作，會在 Google Cloud 專案中建立由 HTTP 觸發的基本 Cloud Run 函式。

Cloud 函式範本

下列範例說明如何在整合中，以不同語言使用 Cloud Functions 工作。

Python

使用現有 Cloud Run 函式設定 Cloud Function 時，請確認函式的 main.py、task.py 和 requirements.txt 來源檔案採用下列格式：

task.py

      # Sample Code:
      # print(event.get('task_string_key'))
      # event.set('task_int_array_key', [456, 789]);
      # event.log('some logging')

      def run(event):
        """Actual cloud function custom logic.
        Args:
          event : event object in main.py that contains all parameters.
        """
        return
    

main.py

      """Un-editable platform wrapper which invokes user code."""
    import traceback

    from flask import json
    from flask import jsonify
    from task import run

    VALUE_NAME = [
        'stringValue', 'intValue', 'doubleValue', 'booleanValue', 'protoValue'
    ]
    ARRAY_VALUE_NAME = {
        'stringArray': 'stringValues',
        'intArray': 'intValues',
        'doubleArray': 'doubleValues',
        'booleanArray': 'booleanValues',
        'protoArray': 'protoValues'
    }
    VALUE_TYPE_URL = 'type.googleapis.com/google.protobuf.Value'
    CLOUD_FUNCTION_EXCEPTION_KEY = 'CloudFunctionException'
    CLOUD_FUNCTION_LOGGING_KEY = 'CloudFunctionLogging'


    class _Event(object):
      """Event object."""

      def __init__(self, json_payload):
        self._event_params = json_payload.get('eventParameters', dict())
        self._task_params = json_payload.get('taskParameters', dict())
        self._log = []
        print('Event param is ' + str(self._event_params))
        print('Task param is ' + str(self._task_params))

      def set(self, key, value):
        """Set the event parameters key-value.

        Args:
          key: parameter key.
          value: parameter value.
        """
        new_param = self._create_param(key, value)
        param = self._get_param_by_key(key)
        if param is None:
          if 'parameters' not in self._event_params:
            self._event_params['parameters'] = []
          self._event_params['parameters'].append(new_param)
        else:
          param['value'] = new_param['value']

      def _create_param(self, key, value):
        """Create a new parameter with given key value pair.

        Args:
          key: parameter key.
          value: parameter value.

        Returns:
          parameter.
        """
        new_param = {}
        new_param['key'] = key
        if isinstance(value, str):
          new_param['value'] = {'stringValue': value}
        elif isinstance(value, int):
          new_param['value'] = {'intValue': value}
        elif isinstance(value, float):
          new_param['value'] = {'doubleValue': value}
        elif isinstance(value, bool):
          new_param['value'] = {'booleanValue': value}
        elif isinstance(value, dict):
          if 'type@' in value:
            new_param['value'] = {'protoValue': value}
          else:
            new_param['value'] = {
                'protoValue': {
                    '@type': 'type.googleapis.com/google.protobuf.Value',
                    'value': value
                }
            }
        elif isinstance(value, list):
          if not value:
            raise RuntimeError('Cannot create a param with empty list')
          if any(not isinstance(val, type(value[0])) for val in value):
            print('Not all elements in the list have the same type')
            new_param['value'] = {
                'protoValue': {
                    '@type': 'type.googleapis.com/google.protobuf.Value',
                    'value': value
                }
            }
          elif isinstance(value[0], str):
            new_param['value'] = {'stringArray': {'stringValues': value}}
          elif isinstance(value[0], int):
            new_param['value'] = {'intArray': {'intValues': value}}
          elif isinstance(value[0], float):
            new_param['value'] = {'doubleArray': {'doubleValues': value}}
          elif isinstance(value[0], bool):
            new_param['value'] = {'booleanArray': {'booleanValues': value}}
          elif isinstance(value[0], dict):
            if all('@type' in val and val['@type'] == value[0]['@type']
                   for val in value):
              new_param['value'] = {'protoArray': {'protoValues': value}}
            else:
              new_param['value'] = {
                  'protoValue': {
                      '@type': 'type.googleapis.com/google.protobuf.Value',
                      'value': value
                  }
              }
          else:
            raise RuntimeError('The type ' + type(value[0]) +
                               ' in the list is not supported')
        else:
          raise RuntimeError('Value ' + str(value) + ' has the type ' +
                             type(value) + ' that is not supported')
        return new_param

      def get(self, key):
        """Get the event parameter value for specified key.

        Args:
          key: parameter key.

        Returns:
          Parameter value.
        """
        param = self._get_param_by_key(key)
        if param is None:
          raise RuntimeError('Can not find param with key ' + key)
        return self._get_param_value(param)

      def _get_param_by_key(self, key):
        """Get the parameter for specified key.

        Args:
          key: parameter key.

        Returns:
          Parameter.
        """
        param = self._get_param_by_key_from_params(key, self._task_params)
        if param is None:
          return self._get_param_by_key_from_params(key, self._event_params)
        value = self._get_param_value(param)
        if isinstance(value, str) and len(value) > 2 and value.startswith(
            '$') and value.endswith('$'):
          return self._get_param_by_key_from_params(value[1:-1], self._event_params)
        return param

      def _get_param_by_key_from_params(self, key, params):
        """Get the parameter for specified key from event parameters.

        Args:
          key: parameter key.
          params: event parameters.

        Returns:
          Parameter.
        """
        if not isinstance(params, dict) or 'parameters' not in params:
          return None
        for param in params['parameters']:
          if param['key'] == key:
            return param
        return None

      def _get_param_value(self, param):
        """Get the parameter value for specified parameter.

        Args:
          param: parameter.

        Returns:
          Parameter value.
        """
        value = param['value']
        if len(value) != 1:
          raise RuntimeError('param does not have size of 1')
        for value_name in VALUE_NAME:
          if value_name in value:
            if value_name == 'protoValue' and value[value_name][
                '@type'] == VALUE_TYPE_URL:
              return value[value_name]['value']
            return value[value_name]
        for array_value_name in ARRAY_VALUE_NAME:
          if array_value_name in value:
            return value[array_value_name][ARRAY_VALUE_NAME[array_value_name]]
        raise RuntimeError('Cannot get value from param ' + str(param))

      def set_error(self):
        """Set the cloud function error to event parameters in order for user to see on IP."""

        self.set(CLOUD_FUNCTION_EXCEPTION_KEY, traceback.format_exc())

      def log(self, message):
        self._log.append(str(message))

      def get_response(self):
        """Get the response that can be returned to IP.

        Returns:
          The response text or any set of values that can be turned into a
          Response object using
          `make_response
          <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
        """
        if self._log:
          self.set(CLOUD_FUNCTION_LOGGING_KEY, self._log)
        res = {
            'eventParameters': self._event_params,
        }
        return jsonify(res)


    def execute_function(request):
      """Entry point of the cloud function.

      Args:
        request (flask.Request): HTTP request object.

      Returns:
        The response text or any set of values that can be turned into a
        Response object using
        `make_response
        <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
      """
      try:
        request_json = request.get_json(silent=True)
        event = _Event(request_json)
        run(event)
      except:
        event.set_error()
      return event.get_response()

    

requirements.txt

    # Function dependencies, for example:
    # package>=version
    

如要進一步瞭解回應格式，請參閱「ValueType」。

Java

下列範例說明如何在整合中，使用 Cloud Functions 工作。請確認回覆內容符合範例中說明的支援 JSON 格式：

private static final Gson gson = new Gson();

@Override
public void service(HttpRequest request, HttpResponse response) throws Exception {
  JsonObject body = gson.fromJson(request.getReader(), JsonObject.class);

  JsonArray resParams = new JsonArray();
  for (JsonElement param: body.getAsJsonObject("eventParameters").getAsJsonArray("parameters")) {
    if (param.getAsJsonObject().get("key").getAsString().equals("input")) {
      JsonObject newParam= new JsonObject();
      newParam.addProperty("key", "input");
      JsonObject value = new JsonObject();
      value.addProperty("stringValue","2");
      newParam.add("value", value);
      resParams.add(newParam);
    } else {
      resParams.add(param);
    }
  }
  JsonObject parameters = new JsonObject();
  parameters.add("parameters", resParams);
  JsonObject res = new JsonObject();
  res.add("eventParameters", parameters);
  System.out.println(res);
  BufferedWriter writer = response.getWriter();
  writer.write(res.toString());
}

如要進一步瞭解回應格式，請參閱「ValueType」。

JavaScript

下列範例說明如何在整合中，使用 Cloud Functions 工作。請確認回覆內容符合範例中說明的支援 JSON 格式：

const functions = require('@google-cloud/functions-framework');

functions.http('execute_function', (req, res) => {
  console.log(JSON.stringify(req.body));
  let response = {"eventParameters":{"parameters":[{"key":"input","value":{"stringValue":"2"}}]}};
  res.send(JSON.stringify(response));
});

如要進一步瞭解回應格式，請參閱「ValueType」。

PHP

下列範例說明如何在整合中，使用 Cloud Functions 工作。請確認回覆內容符合範例中說明的支援 JSON 格式：

use Psr\Http\Message\ServerRequestInterface;
function execute_function(ServerRequestInterface $request)
{
  return '{"eventParameters":{"parameters":[{"key":"input","value":{"stringValue":"2"}}]}}';
}

如要進一步瞭解回應格式，請參閱「ValueType」。

編輯 Cloud Functions 工作

Application Integration 會根據版本類型，將您導向適當的 Google Cloud 控制台頁面，以編輯 Cloud Run 函式。

Cloud Functions (第 1 代)

如要編輯使用 Cloud Functions (第 1 代) 版本設定的 Cloud Functions 工作，請按照下列步驟操作：

1. 在工作設定窗格中，按一下「開啟 Cloud Function」。

   系統會將您導向 Google Cloud console中的「Cloud Functions (第 1 代) > 函式詳細資料」頁面。

2. 按一下 [編輯]。
3. 在「編輯函式」頁面中，您可以在「設定」步驟編輯 Cloud Functions 的預設設定。 詳情請參閱「設定 Cloud Functions」。
4. 按一下「Next」(下一步)，前往「Code」(程式碼) 步驟，編輯 Cloud 函式的原始碼。

   根據預設，Cloud 函式包含下列來源檔案：

   • main.py：這個檔案包含初始化程式碼，可從整合服務執行 Cloud Function。
     注意：請勿編輯或修改 main.py 檔案。
   • task.py：這個檔案包含 Cloud Function 的可執行程式碼。 在 run(event) 函式中編寫指令碼。這個函式會在 Cloud Functions 工作執行時呼叫。 main.py 檔案中的 event 物件包含所有工作參數。

     如要瞭解如何在指令碼中使用整合層級定義的變數，請參閱「存取整合變數」。

5. 按一下 [Deploy] (部署)。

Cloud Run 函式

如要編輯使用 Cloud Functions (最新一代) 版本設定的 Cloud Functions 工作，請按照下列步驟操作：

1. 在工作設定窗格中，按一下「開啟 Cloud Function」。

   系統會將您重新導向至 Google Cloud console中的「Cloud Run functions > Service details」(Cloud Run 函式 > 服務詳細資料) 頁面。

2. 在「來源」分頁中，按一下「編輯來源」，即可編輯 Cloud Run 函式的原始碼檔案。

   根據預設，Cloud Run 函式包含下列來源檔案：

   • main.py：這個檔案包含初始化程式碼，可從整合服務執行 Cloud Functions。
     注意：請勿編輯或修改 main.py 檔案。
   • task.py：這個檔案包含 Cloud Functions 的可執行程式碼。 在 run(event) 函式中編寫指令碼。這項函式會在 Cloud Run 函式工作執行時呼叫。 main.py 檔案中的 event 物件包含所有工作參數。

     如要瞭解如何在指令碼中使用整合層級定義的變數，請參閱「存取整合變數」。

3. 按一下「儲存並重新部署」。

存取整合作業變數

如要在 Cloud Functions 中存取整合變數，請將變數做為工作參數傳遞至 Cloud Functions 工作。工作參數是鍵值組，其中「鍵」是 Cloud 函式來源檔案中使用的參照變數名稱，「值」則是參照變數指向的對應整合變數名稱。 您可以在工作設定窗格的「工作參數」部分新增一或多個工作參數。

下列方法用於從 Cloud Function 存取整合變數：

• set：將值寫入變數。
• get：讀取變數的值。

舉例來說，如果您有名為 EmployeeName 的整合變數，想在 Cloud Functions 來源檔案中使用，請定義下列工作參數：

• 金鑰：EmployeeKey
• 價值：EmployeeName

下列範例指令碼顯示如何使用 set 和 get 函式存取已定義的整合變數。

def run(event):  
  # Read the integration variable EmployeeName using the reference variable EmployeeKey
  value = event.get('EmployeeKey');
  # Change the integration variable EmployeeName value using the reference variable EmployeeKey
  event.set('EmployeeKey' , 'XYZ');
  # The new value of the integration variable is retained throughout the Cloud Function task.
  return

錯誤處理策略

工作錯誤處理策略會指定工作因暫時性錯誤而失敗時，系統應採取的動作。如要瞭解如何使用錯誤處理策略，以及不同類型的錯誤處理策略，請參閱「錯誤處理策略」。

「服務水準協議」排除條款

Cloud Functions 工作會依附 Google Cloud Functions 產品。 由於這項依附元件是 Application Integration 的外部元件，因此凡是因 Cloud Functions 工作失敗而導致 active 整合執行失敗，都排除在 Application Integration 服務水準協議 (SLA) 條款及細則之外。

配額與限制

如要瞭解 Cloud Run functions 和 Cloud Functions (第 1 代) 的配額和限制，請參閱設定比較。

後續步驟

• 瞭解所有工作和觸發條件。
• 瞭解如何測試及發布整合項目。
• 瞭解 錯誤處理。
• 瞭解整合執行記錄。