-
Notifications
You must be signed in to change notification settings - Fork 623
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[ISSUE #4420] Add Feishu Sink connector #4522
Changes from 10 commits
fae2324
aff39ba
1f25a26
edc38d5
d85deec
dc1af2f
3123e6b
4562695
45bb953
24abe12
0cf93d9
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
/* | ||
* 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. | ||
*/ | ||
|
||
List feishu = [ | ||
"com.larksuite.oapi:oapi-sdk:$feishu_version", | ||
"com.github.rholder:guava-retrying:$guava_retrying_version", | ||
"org.apache.httpcomponents:httpclient", | ||
project(":eventmesh-common") | ||
] | ||
|
||
dependencies { | ||
api project(":eventmesh-openconnect:eventmesh-openconnect-java") | ||
implementation feishu | ||
compileOnly 'org.projectlombok:lombok' | ||
annotationProcessor 'org.projectlombok:lombok' | ||
|
||
testImplementation "org.mockito:mockito-core" | ||
testImplementation "org.mockito:mockito-junit-jupiter" | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,18 @@ | ||
# | ||
# 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. | ||
# | ||
feishu_version=2.0.28 | ||
guava_retrying_version=2.0.0 |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
/* | ||
* 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. | ||
*/ | ||
|
||
package org.apache.eventmesh.connector.feishu.server; | ||
|
||
import org.apache.eventmesh.connector.feishu.sink.connector.FeishuSinkConnector; | ||
import org.apache.eventmesh.openconnect.Application; | ||
|
||
import lombok.extern.slf4j.Slf4j; | ||
|
||
@Slf4j | ||
public class FeishuConnectServer { | ||
Check warning on line 26 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/server/FeishuConnectServer.java
|
||
|
||
public static void main(String[] args) throws Exception { | ||
|
||
Application feishuSinkApp = new Application(); | ||
feishuSinkApp.run(FeishuSinkConnector.class); | ||
} | ||
Check warning on line 32 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/server/FeishuConnectServer.java
|
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,31 @@ | ||
/* | ||
* 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. | ||
*/ | ||
|
||
package org.apache.eventmesh.connector.feishu.sink.config; | ||
|
||
import org.apache.eventmesh.openconnect.api.config.SinkConfig; | ||
|
||
import lombok.Data; | ||
import lombok.EqualsAndHashCode; | ||
|
||
@Data | ||
@EqualsAndHashCode(callSuper = true) | ||
public class FeishuSinkConfig extends SinkConfig { | ||
|
||
public SinkConnectorConfig connectorConfig; | ||
|
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,34 @@ | ||
/* | ||
* 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. | ||
*/ | ||
|
||
package org.apache.eventmesh.connector.feishu.sink.config; | ||
|
||
import lombok.Data; | ||
|
||
@Data | ||
public class SinkConnectorConfig { | ||
|
||
private String connectorName; | ||
Check warning on line 25 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/config/SinkConnectorConfig.java
|
||
|
||
private String receiveId; | ||
|
||
private String appId; | ||
Check warning on line 29 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/config/SinkConnectorConfig.java
|
||
|
||
private String appSecret; | ||
Check warning on line 31 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/config/SinkConnectorConfig.java
|
||
|
||
private String receiveIdType; | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,149 @@ | ||
/* | ||
* 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. | ||
*/ | ||
|
||
package org.apache.eventmesh.connector.feishu.sink.connector; | ||
|
||
import static org.apache.eventmesh.common.Constants.FEISHU_CONTENT; | ||
import static org.apache.eventmesh.common.Constants.FEISHU_MSG_TYPE; | ||
import static org.apache.eventmesh.common.Constants.FEISHU_RECEIVE_ID; | ||
import static org.apache.eventmesh.common.Constants.FEISHU_SEND_MESSAGE_API; | ||
import static org.apache.eventmesh.common.Constants.FEISHU_UUID; | ||
|
||
import org.apache.eventmesh.connector.feishu.sink.config.FeishuSinkConfig; | ||
import org.apache.eventmesh.connector.feishu.sink.config.SinkConnectorConfig; | ||
import org.apache.eventmesh.openconnect.api.config.Config; | ||
import org.apache.eventmesh.openconnect.api.connector.ConnectorContext; | ||
import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext; | ||
import org.apache.eventmesh.openconnect.api.sink.Sink; | ||
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord; | ||
|
||
import java.util.HashMap; | ||
import java.util.List; | ||
import java.util.Map; | ||
import java.util.UUID; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.atomic.AtomicReference; | ||
|
||
import com.github.rholder.retry.Attempt; | ||
import com.github.rholder.retry.RetryListener; | ||
import com.github.rholder.retry.Retryer; | ||
import com.github.rholder.retry.RetryerBuilder; | ||
import com.github.rholder.retry.StopStrategies; | ||
import com.github.rholder.retry.WaitStrategies; | ||
import com.lark.oapi.Client; | ||
import com.lark.oapi.core.response.RawResponse; | ||
import com.lark.oapi.core.token.AccessTokenType; | ||
import com.lark.oapi.service.im.v1.enums.MsgTypeEnum; | ||
import com.lark.oapi.service.im.v1.model.ext.MessageText; | ||
|
||
import lombok.extern.slf4j.Slf4j; | ||
|
||
@Slf4j | ||
public class FeishuSinkConnector implements Sink { | ||
|
||
private FeishuSinkConfig sinkConfig; | ||
|
||
private Client feishuClient; | ||
|
||
private static final int MAX_RETRY_TIME = 3; | ||
|
||
private static final int FIXED_WAIT_SECOND = 1; | ||
|
||
private final Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder().retryIfException().retryIfResult(res -> !res) | ||
.withWaitStrategy(WaitStrategies.fixedWait(FIXED_WAIT_SECOND, TimeUnit.SECONDS)) | ||
.withStopStrategy(StopStrategies.stopAfterAttempt(MAX_RETRY_TIME)).withRetryListener(new RetryListener() { | ||
|
||
@Override | ||
public <V> void onRetry(Attempt<V> attempt) { | ||
long times = attempt.getAttemptNumber(); | ||
log.warn("retry invoke http,times={}", times); | ||
} | ||
}).build(); | ||
|
||
@Override | ||
public Class<? extends Config> configClass() { | ||
return FeishuSinkConfig.class; | ||
Check warning on line 79 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
} | ||
|
||
@Override | ||
public void init(Config config) throws Exception { | ||
// init config for feishu sink connector | ||
this.sinkConfig = (FeishuSinkConfig) config; | ||
doInit(); | ||
} | ||
Check warning on line 87 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
|
||
@Override | ||
public void init(ConnectorContext connectorContext) throws Exception { | ||
// init config for feishu sink connector | ||
SinkConnectorContext sinkConnectorContext = (SinkConnectorContext) connectorContext; | ||
this.sinkConfig = (FeishuSinkConfig) sinkConnectorContext.getSinkConfig(); | ||
doInit(); | ||
} | ||
Check warning on line 95 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
|
||
private void doInit() { | ||
this.feishuClient = Client.newBuilder(this.getConfig().getAppId(), this.getConfig().getAppSecret()).requestTimeout(3, TimeUnit.SECONDS) | ||
.build(); | ||
} | ||
Check warning on line 100 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
|
||
@Override | ||
public void start() { | ||
} | ||
|
||
@Override | ||
public void commit(ConnectRecord record) { | ||
|
||
} | ||
Check warning on line 109 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
|
||
@Override | ||
public String name() { | ||
return this.sinkConfig.getConnectorConfig().getConnectorName(); | ||
Check warning on line 113 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
} | ||
|
||
@Override | ||
public void stop() { | ||
} | ||
Check warning on line 118 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
|
||
@Override | ||
public void put(List<ConnectRecord> sinkRecords) { | ||
SinkConnectorConfig connectorConfig = getConfig(); | ||
try { | ||
for (ConnectRecord connectRecord : sinkRecords) { | ||
AtomicReference<RawResponse> response = new AtomicReference<>(); | ||
retryer.call(() -> { | ||
Map<String, Object> body = new HashMap<>(); | ||
body.put(FEISHU_RECEIVE_ID, connectorConfig.getReceiveId()); | ||
body.put(FEISHU_CONTENT, MessageText.newBuilder().text(connectRecord.getData().toString()).build()); | ||
body.put(FEISHU_MSG_TYPE, MsgTypeEnum.MSG_TYPE_TEXT.getValue()); | ||
body.put(FEISHU_UUID, UUID.randomUUID().toString()); | ||
response.set(feishuClient.post(FEISHU_SEND_MESSAGE_API + connectorConfig.getReceiveIdType(), body, AccessTokenType.Tenant)); | ||
if (response.get().getStatusCode() != 200) { | ||
log.error("request feishu open api err{}", new String(response.get().getBody())); | ||
return false; | ||
Check warning on line 135 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
} | ||
return true; | ||
}); | ||
} | ||
} catch (Exception e) { | ||
log.error("failed to put message to feishu", e); | ||
Check warning on line 141 in eventmesh-connectors/eventmesh-connector-feishu/src/main/java/org/apache/eventmesh/connector/feishu/sink/connector/FeishuSinkConnector.java
|
||
} | ||
} | ||
|
||
public SinkConnectorConfig getConfig() { | ||
return this.sinkConfig.getConnectorConfig(); | ||
} | ||
|
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,30 @@ | ||
# | ||
# 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. | ||
# | ||
|
||
pubSubConfig: | ||
meshAddress: 127.0.0.1:10000 | ||
subject: TopicTest | ||
idc: FT | ||
env: PRD | ||
group: feishuSink | ||
appId: 5031 | ||
userName: feishuSinkUser | ||
passWord: feishuPassWord | ||
connectorConfig: | ||
connectorName: feishuSink | ||
reciveId: reciveIdValue | ||
reciveType: open_id | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The
reciveId
andrecoverType
fields in thesink-config.yml
are different from the field names inSinkConnectorConfig
, which will cause the configuration to fail to be imported correctly.@SunnyBoy-WYH 配置文件中reciveId和reciveType字段与SinkConnectorConfig中字段名不同,这将导致配置无法正确导入。
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Are you willing to fix this bug? @hhuang1231 .
Additionally, this bug makes me worried about the correctness and effectiveness of this Connector in the absence of this bug.
您是否愿意修改这个bug?@hhuang1231 。另外,该bug也让我有点担心,当没有该bug时,该Connector的正确性和有效性。
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@pandaapo ok, I'll try to perfect it and make it work.